package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"iter"
	"log/slog"
	"time"

	"git.cloudinit.dev/coreci/orca/internal/model"
)

// watchInterval is the poll cadence used by JobRepo.Watch and NodeRepo.Watch.
// It is an unexported package var (default 1s) so tests can override it to a
// small value for deterministic assertions (D-043). Do not change it from
// production code paths.
var watchInterval = 1 * time.Second

type JobRepo struct {
	db *sql.DB
}

func NewJobRepo(db *sql.DB) *JobRepo {
	return &JobRepo{db: db}
}

func (r *JobRepo) Insert(ctx context.Context, j *model.Job) error {
	if j.CreatedAt.IsZero() {
		j.CreatedAt = time.Now().UTC()
	}
	if j.Status == "" {
		j.Status = model.JobStatusPending
	}
	_, err := r.db.ExecContext(ctx,
		`INSERT INTO jobs (id, name, spec, status, exit_code, created_at) VALUES (?, ?, ?, ?, ?, ?)`,
		j.ID, j.Name, j.Spec, string(j.Status), j.ExitCode, j.CreatedAt)
	if err != nil {
		return fmt.Errorf("insert job: %w", err)
	}
	return nil
}

func (r *JobRepo) Get(ctx context.Context, id string) (*model.Job, error) {
	row := r.db.QueryRowContext(ctx,
		`SELECT id, name, spec, status, exit_code, created_at, started_at, ended_at FROM jobs WHERE id = ?`, id)
	return scanJob(row)
}

func (r *JobRepo) List(ctx context.Context) ([]*model.Job, error) {
	rows, err := r.db.QueryContext(ctx,
		`SELECT id, name, spec, status, exit_code, created_at, started_at, ended_at FROM jobs ORDER BY created_at DESC`)
	if err != nil {
		return nil, fmt.Errorf("list jobs: %w", err)
	}
	defer rows.Close()
	var jobs []*model.Job
	for rows.Next() {
		j, err := scanJob(rows)
		if err != nil {
			return nil, err
		}
		jobs = append(jobs, j)
	}
	return jobs, rows.Err()
}

// Watch yields the full snapshot of jobs on a watchInterval ticker until ctx
// is cancelled or the consumer stops pulling (yield returns false). It does
// not spawn a goroutine; the polling loop runs inline in the caller's
// goroutine via the range-over-func pull protocol (D-032).
//
// Each tick re-runs the List query and yields one []*model.Job snapshot
// containing ALL rows for that tick (G-001). The first yield happens
// immediately before the first ticker wait, so the consumer sees the initial
// state with no watchInterval delay (G-002). Transient query/scan errors are
// logged via slog.Default().Warn and the loop continues to the next tick
// rather than terminating the stream (D-034 lite). The ticker is stopped and
// rows are closed on every exit path (ctx.Done, yield==false, scan error).
func (r *JobRepo) Watch(ctx context.Context) iter.Seq[[]*model.Job] {
	return func(yield func([]*model.Job) bool) {
		ticker := time.NewTicker(watchInterval)
		defer ticker.Stop()
		for {
			rows, err := r.db.QueryContext(ctx,
				`SELECT id, name, spec, status, exit_code, created_at, started_at, ended_at FROM jobs ORDER BY created_at DESC`)
			if err != nil {
				slog.Default().Warn("watch jobs: query failed", "error", err)
				// fall through to the select to wait for the next tick
			} else {
				snapshot := make([]*model.Job, 0)
				for rows.Next() {
					j, scanErr := scanJob(rows)
					if scanErr != nil {
						slog.Default().Warn("watch jobs: scan failed", "error", scanErr)
						continue
					}
					snapshot = append(snapshot, j)
				}
				rows.Close()
				if !yield(snapshot) {
					return // consumer stopped pulling
				}
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}
}

func (r *JobRepo) UpdateStatus(ctx context.Context, id string, status model.JobStatus, exitCode int) error {
	now := time.Now().UTC()
	var startedAt, endedAt *time.Time
	switch status {
	case model.JobStatusRunning:
		startedAt = &now
	case model.JobStatusComplete, model.JobStatusFailed, model.JobStatusStopped:
		endedAt = &now
	}
	_, err := r.db.ExecContext(ctx,
		`UPDATE jobs SET status = ?, exit_code = ?, started_at = COALESCE(?, started_at), ended_at = COALESCE(?, ended_at) WHERE id = ?`,
		string(status), exitCode, startedAt, endedAt, id)
	if err != nil {
		return fmt.Errorf("update job: %w", err)
	}
	return nil
}

func scanJob(s scanner) (*model.Job, error) {
	var (
		j         model.Job
		status    string
		startedAt sql.NullTime
		endedAt   sql.NullTime
	)
	err := s.Scan(&j.ID, &j.Name, &j.Spec, &status, &j.ExitCode, &j.CreatedAt, &startedAt, &endedAt)
	if err == sql.ErrNoRows {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("scan job: %w", err)
	}
	j.Status = model.JobStatus(status)
	if startedAt.Valid {
		j.StartedAt = &startedAt.Time
	}
	if endedAt.Valid {
		j.EndedAt = &endedAt.Time
	}
	return &j, nil
}

type TaskRepo struct {
	db *sql.DB
}

func NewTaskRepo(db *sql.DB) *TaskRepo {
	return &TaskRepo{db: db}
}

func (r *TaskRepo) Insert(ctx context.Context, t *model.Task) error {
	if t.CreatedAt.IsZero() {
		t.CreatedAt = time.Now().UTC()
	}
	if t.Status == "" {
		t.Status = model.TaskStatusPending
	}
	argsJSON, _ := json.Marshal(t.Args)
	envJSON, _ := json.Marshal(t.Env)
	_, err := r.db.ExecContext(ctx,
		`INSERT INTO tasks (id, job_id, command, args, env, pid, exit_code, status, created_at, stdout, stderr)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		t.ID, t.JobID, t.Command, string(argsJSON), string(envJSON),
		t.PID, t.ExitCode, string(t.Status), t.CreatedAt, t.Stdout, t.Stderr)
	if err != nil {
		return fmt.Errorf("insert task: %w", err)
	}
	return nil
}

func (r *TaskRepo) Get(ctx context.Context, id string) (*model.Task, error) {
	row := r.db.QueryRowContext(ctx,
		`SELECT id, job_id, command, args, env, pid, exit_code, status, created_at, started_at, ended_at, stdout, stderr FROM tasks WHERE id = ?`, id)
	return scanTask(row)
}

func (r *TaskRepo) ListByJob(ctx context.Context, jobID string) ([]*model.Task, error) {
	rows, err := r.db.QueryContext(ctx,
		`SELECT id, job_id, command, args, env, pid, exit_code, status, created_at, started_at, ended_at, stdout, stderr FROM tasks WHERE job_id = ? ORDER BY created_at ASC`, jobID)
	if err != nil {
		return nil, fmt.Errorf("list tasks: %w", err)
	}
	defer rows.Close()
	var tasks []*model.Task
	for rows.Next() {
		t, err := scanTask(rows)
		if err != nil {
			return nil, err
		}
		tasks = append(tasks, t)
	}
	return tasks, rows.Err()
}

func (r *TaskRepo) UpdateRunning(ctx context.Context, id string, pid int) error {
	now := time.Now().UTC()
	_, err := r.db.ExecContext(ctx,
		`UPDATE tasks SET pid = ?, status = ?, started_at = ? WHERE id = ?`,
		pid, string(model.TaskStatusRunning), now, id)
	if err != nil {
		return fmt.Errorf("update task running: %w", err)
	}
	return nil
}

func (r *TaskRepo) UpdateDone(ctx context.Context, id string, exitCode int, stdout, stderr string) error {
	now := time.Now().UTC()
	status := model.TaskStatusComplete
	if exitCode != 0 {
		status = model.TaskStatusFailed
	}
	_, err := r.db.ExecContext(ctx,
		`UPDATE tasks SET status = ?, exit_code = ?, ended_at = ?, stdout = ?, stderr = ? WHERE id = ?`,
		string(status), exitCode, now, stdout, stderr, id)
	if err != nil {
		return fmt.Errorf("update task done: %w", err)
	}
	return nil
}

func (r *TaskRepo) UpdateKilled(ctx context.Context, id string) error {
	now := time.Now().UTC()
	_, err := r.db.ExecContext(ctx,
		`UPDATE tasks SET status = ?, ended_at = ? WHERE id = ?`,
		string(model.TaskStatusKilled), now, id)
	if err != nil {
		return fmt.Errorf("update task killed: %w", err)
	}
	return nil
}

// ListRecent returns up to limit tasks ordered by created_at DESC.
// Used by the API to expose recent activity without a job filter.
func (r *TaskRepo) ListRecent(ctx context.Context, limit int) ([]*model.Task, error) {
	if limit <= 0 {
		limit = 100
	}
	rows, err := r.db.QueryContext(ctx,
		`SELECT id, job_id, command, args, env, pid, exit_code, status, created_at, started_at, ended_at, stdout, stderr FROM tasks ORDER BY created_at DESC LIMIT ?`, limit)
	if err != nil {
		return nil, fmt.Errorf("list tasks recent: %w", err)
	}
	defer rows.Close()
	var tasks []*model.Task
	for rows.Next() {
		t, err := scanTask(rows)
		if err != nil {
			return nil, err
		}
		tasks = append(tasks, t)
	}
	return tasks, rows.Err()
}

var _ = errors.New
var _ = json.Marshal

func scanTask(s scanner) (*model.Task, error) {
	var (
		t         model.Task
		status    string
		argsJSON  string
		envJSON   string
		startedAt sql.NullTime
		endedAt   sql.NullTime
	)
	err := s.Scan(&t.ID, &t.JobID, &t.Command, &argsJSON, &envJSON,
		&t.PID, &t.ExitCode, &status, &t.CreatedAt, &startedAt, &endedAt, &t.Stdout, &t.Stderr)
	if err == sql.ErrNoRows {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("scan task: %w", err)
	}
	t.Status = model.TaskStatus(status)
	if startedAt.Valid {
		t.StartedAt = &startedAt.Time
	}
	if endedAt.Valid {
		t.EndedAt = &endedAt.Time
	}
	_ = json.Unmarshal([]byte(argsJSON), &t.Args)
	_ = json.Unmarshal([]byte(envJSON), &t.Env)
	return &t, nil
}
