package hub

import (
	"context"
	"strconv"
	"strings"

	"github.com/shcizo/package-updater/internal/master/events"
	"github.com/shcizo/package-updater/internal/master/model"
	"github.com/shcizo/package-updater/internal/protocol"
)

// StartJob sends job.start over exactly the connection r names. It returns
// model.ErrAgentOffline if that connection is gone, was replaced by a newer
// one, or has not finished its hello.
func (h *Hub) StartJob(r model.Route, jobID int64, start protocol.JobStart) error {
	h.mu.Lock()
	c := h.conns[r.AgentID]
	usable := c != nil && c.id == r.ConnID && c.hello != nil
	h.mu.Unlock()
	if !usable {
		return model.ErrAgentOffline
	}
	env, err := protocol.EncodeJob(protocol.TypeJobStart, strconv.FormatInt(jobID, 10), start)
	if err != nil {
		return err
	}
	// A job whose write later fails stays queued; the agent's next hello
	// settles it as lost.
	if !c.send(env) {
		return model.ErrAgentOffline
	}
	return nil
}

// parseJobIDs converts the agent's job IDs to integers, calling skip for each
// one that is not a number.
func parseJobIDs(ids []string, skip func(id string)) []int64 {
	out := make([]int64, 0, len(ids))
	for _, s := range ids {
		id, err := strconv.ParseInt(s, 10, 64)
		if err != nil {
			skip(s)
			continue
		}
		out = append(out, id)
	}
	return out
}

// cleanOutput drops NUL bytes, which Postgres text columns reject. Without
// this a result containing one could never be stored and would be resent
// forever.
func cleanOutput(s string) string { return strings.ReplaceAll(s, "\x00", "") }

func (s *session) handleJobProgress(ctx context.Context, env protocol.Envelope) error {
	h := s.hub
	v, err := protocol.Decode[protocol.JobProgress](env)
	if err != nil {
		return err
	}
	id, err := strconv.ParseInt(env.ID, 10, 64)
	if err != nil {
		h.log.Warn("job_progress_ignored", "agent", s.agent.Name, "job_id", env.ID, "reason", "not a number")
		return nil
	}
	applied, err := h.store.JobProgress(ctx, s.agent.ID, id, cleanOutput(v.Output), h.now())
	if err != nil {
		return err
	}
	if applied {
		h.pub.Publish(events.Jobs)
		h.pub.Publish(events.JobTopic(id))
	}
	return nil
}

func (s *session) handleJobResult(ctx context.Context, env protocol.Envelope) error {
	h := s.hub
	v, err := protocol.Decode[protocol.JobResult](env)
	if err != nil {
		return err
	}
	ack, err := protocol.EncodeJob(protocol.TypeJobAck, env.ID, protocol.JobAck{})
	if err != nil {
		return err
	}
	id, err := strconv.ParseInt(env.ID, 10, 64)
	if err != nil {
		h.log.Warn("job_result_ignored", "agent", s.agent.Name, "job_id", env.ID, "reason", "not a number")
		s.conn.send(ack)
		return nil
	}

	fin := model.JobFinish{
		Status:          model.JobStatus(v.Status),
		Ref:             v.Ref,
		Error:           v.Error,
		Output:          cleanOutput(v.Output),
		OutputTruncated: v.OutputTruncated,
		StartedAt:       v.StartedAt,
		FinishedAt:      v.FinishedAt,
	}
	if v.Status != protocol.ResultSucceeded && v.Status != protocol.ResultFailed {
		fin.Status = model.JobFailed
		fin.Ref = ""
		fin.Error = "agent sent invalid status " + strconv.Quote(v.Status)
	}
	if fin.StartedAt.IsZero() {
		fin.StartedAt = h.now()
	}
	if fin.FinishedAt.IsZero() {
		fin.FinishedAt = h.now()
	}

	// A store error closes the connection without an ack, so the agent
	// resends the result after it reconnects.
	applied, err := h.store.FinishJob(ctx, s.agent.ID, id, fin)
	if err != nil {
		return err
	}
	if applied {
		h.pub.Publish(events.Jobs)
		h.pub.Publish(events.JobTopic(id))
		if fin.Status == model.JobSucceeded {
			h.pub.Publish(events.Services)
		}
	} else {
		h.log.Warn("job_result_ignored", "agent", s.agent.Name, "job_id", env.ID, "reason", "no such active job for this agent")
	}
	s.conn.send(ack)
	return nil
}