package db

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"tangled.org/core/log"
	"tangled.org/core/sqlite"
)

var (
	ErrNotFound          = errors.New("not found")
	ErrConflict          = errors.New("idempotency conflict")
	ErrInvalidTransition = errors.New("invalid status transition")
)

type InvalidTransitionError struct {
	Status JobStatus
}

func (e *InvalidTransitionError) Error() string {
	return fmt.Sprintf("%s: cannot retry job with status %q (must be failed or authorization_required)", ErrInvalidTransition, e.Status)
}

func (e *InvalidTransitionError) Is(target error) bool { return target == ErrInvalidTransition }

type DB struct {
	*sql.DB
	logger *slog.Logger

	MaxActivePerOwner int
}

func Make(ctx context.Context, dbPath string) (*DB, error) {
	logger := log.SubLogger(log.FromContext(ctx), "db")

	db, err := sqlite.Open(dbPath)
	if err != nil {
		return nil, fmt.Errorf("opening sqlite database: %w", err)
	}

	conn, err := db.Conn(ctx)
	if err != nil {
		return nil, fmt.Errorf("getting sqlite connection: %w", err)
	}
	defer conn.Close()

	if _, err := conn.ExecContext(ctx, schema); err != nil {
		return nil, fmt.Errorf("applying schema: %w", err)
	}
	if err := addColumnIfMissing(ctx, conn, "jobs", "capability_token", "text"); err != nil {
		return nil, fmt.Errorf("migrating jobs capability token: %w", err)
	}
	if err := addColumnIfMissing(ctx, conn, "jobs", "description", "text not null default ''"); err != nil {
		return nil, fmt.Errorf("migrating jobs description: %w", err)
	}
	if _, err := conn.ExecContext(ctx, "create unique index if not exists idx_jobs_capability_token on jobs(capability_token) where capability_token is not null"); err != nil {
		return nil, fmt.Errorf("indexing jobs capability token: %w", err)
	}

	return &DB{DB: db, logger: logger}, nil
}

const schema = `
create table if not exists batches (
	id text primary key,
	owner_did text not null,
	request_id text not null,
	request_digest text not null,
	encrypted_token text,
	credential_expires_at text,
	created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
	updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
	unique(owner_did, request_id)
);

create index if not exists idx_batches_owner_created on batches(owner_did, created_at desc);
create index if not exists idx_batches_cred_exp on batches(credential_expires_at);

create table if not exists jobs (
	id integer primary key autoincrement,
	batch_id text not null references batches(id) on delete cascade,
	repo_did text not null,
	name text not null,
	knot_did text not null,
	source_url text not null,
	description text not null default '',
	private integer not null default 0,
	status text not null default 'queued',
	attempts integer not null default 0,
	next_attempt_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
	error text,
	capability_token text,
	created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
	updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);

create index if not exists idx_jobs_batch_id on jobs(batch_id);
create index if not exists idx_jobs_status_due on jobs(status, next_attempt_at);
create table if not exists oauth_sessions (
	owner_did text primary key,
	session_id text not null,
	encrypted_data text not null,
	updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);

create table if not exists oauth_requests (
	state text primary key,
	encrypted_data text not null,
	created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
);
`

func addColumnIfMissing(ctx context.Context, conn *sql.Conn, table, column, definition string) error {
	rows, err := conn.QueryContext(ctx, "pragma table_info("+table+")")
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var cid int
		var name, typ string
		var notNull, pk int
		var defaultValue sql.NullString
		if err := rows.Scan(&cid, &name, &typ, &notNull, &defaultValue, &pk); err != nil {
			return err
		}
		if name == column {
			return nil
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, "alter table "+table+" add column "+column+" "+definition)
	return err
}

type CreateBatchInput struct {
	ID             string
	OwnerDid       string
	RequestID      string
	RequestDigest  string
	EncryptedToken *string
	Jobs           []CreateJobInput
}

type CreateJobInput struct {
	Name        string
	KnotDid     string
	SourceURL   string
	Description string
	Private     bool
	RepoDid     string
}

func (d *DB) CreateBatch(ctx context.Context, in CreateBatchInput) (*Batch, []Job, bool, error) {
	tx, err := d.BeginTx(ctx, nil)
	if err != nil {
		return nil, nil, false, err
	}
	defer tx.Rollback()

	var existing Batch
	var existingToken sql.NullString
	var credExpStr sql.NullString
	var createdAtStr, updatedAtStr string
	err = tx.QueryRowContext(ctx, `
		select id, owner_did, request_id, request_digest, encrypted_token, credential_expires_at, created_at, updated_at
		from batches where owner_did = ? and request_id = ?
	`, in.OwnerDid, in.RequestID).Scan(
		&existing.ID,
		&existing.OwnerDid,
		&existing.RequestID,
		&existing.RequestDigest,
		&existingToken,
		&credExpStr,
		&createdAtStr,
		&updatedAtStr,
	)

	if err == nil {
		if existing.RequestDigest != in.RequestDigest {
			return nil, nil, false, ErrConflict
		}
		if existingToken.Valid {
			existing.EncryptedToken = &existingToken.String
		}
		if credExpStr.Valid {
			t, _ := time.Parse(time.RFC3339, credExpStr.String)
			existing.CredentialExpiresAt = &t
		}
		existing.CreatedAt, _ = time.Parse(time.RFC3339, createdAtStr)
		existing.UpdatedAt, _ = time.Parse(time.RFC3339, updatedAtStr)

		jobs, err := jobsForBatch(ctx, tx, existing.ID)
		if err != nil {
			return nil, nil, false, err
		}
		return &existing, jobs, true, nil
	} else if !errors.Is(err, sql.ErrNoRows) {
		return nil, nil, false, fmt.Errorf("querying existing batch: %w", err)
	}

	now := time.Now().UTC()
	nowStr := now.Format(time.RFC3339)

	var tokenArg any
	var credExpArg any
	var credExpiresAt *time.Time
	if in.EncryptedToken != nil {
		tokenArg = *in.EncryptedToken
		exp := now.Add(1 * time.Hour)
		credExpiresAt = &exp
		credExpArg = exp.Format(time.RFC3339)
	}

	_, err = tx.ExecContext(ctx, `
		insert into batches (id, owner_did, request_id, request_digest, encrypted_token, credential_expires_at, created_at, updated_at)
		values (?, ?, ?, ?, ?, ?, ?, ?)
	`, in.ID, in.OwnerDid, in.RequestID, in.RequestDigest, tokenArg, credExpArg, nowStr, nowStr)
	if err != nil {
		return nil, nil, false, fmt.Errorf("inserting batch: %w", err)
	}

	createdJobs := make([]Job, 0, len(in.Jobs))
	for _, j := range in.Jobs {
		privInt := 0
		if j.Private {
			privInt = 1
		}
		res, err := tx.ExecContext(ctx, `
			insert into jobs (batch_id, repo_did, name, knot_did, source_url, description, private, status, attempts, next_attempt_at, created_at, updated_at)
			values (?, ?, ?, ?, ?, ?, ?, 'queued', 0, ?, ?, ?)
		`, in.ID, j.RepoDid, j.Name, j.KnotDid, j.SourceURL, j.Description, privInt, nowStr, nowStr, nowStr)
		if err != nil {
			return nil, nil, false, fmt.Errorf("inserting job: %w", err)
		}
		jobID, err := res.LastInsertId()
		if err != nil {
			return nil, nil, false, fmt.Errorf("getting last insert id: %w", err)
		}
		createdJobs = append(createdJobs, Job{
			ID:            jobID,
			BatchID:       in.ID,
			RepoDid:       j.RepoDid,
			Name:          j.Name,
			KnotDid:       j.KnotDid,
			SourceURL:     j.SourceURL,
			Description:   j.Description,
			Private:       j.Private,
			Status:        StatusQueued,
			Attempts:      0,
			NextAttemptAt: now,
			CreatedAt:     now,
			UpdatedAt:     now,
		})
	}

	if err := tx.Commit(); err != nil {
		return nil, nil, false, fmt.Errorf("committing transaction: %w", err)
	}

	batch := &Batch{
		ID:                  in.ID,
		OwnerDid:            in.OwnerDid,
		RequestID:           in.RequestID,
		RequestDigest:       in.RequestDigest,
		EncryptedToken:      in.EncryptedToken,
		CredentialExpiresAt: credExpiresAt,
		CreatedAt:           now,
		UpdatedAt:           now,
	}

	return batch, createdJobs, false, nil
}

type querier interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}

func mapSlice[T, U any](items []T, f func(T) U) []U {
	out := make([]U, len(items))
	for i, it := range items {
		out[i] = f(it)
	}
	return out
}

func jobsForBatches(ctx context.Context, q querier, batchIDs []string) (map[string][]Job, error) {
	if len(batchIDs) == 0 {
		return map[string][]Job{}, nil
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(batchIDs)), ",")
	args := mapSlice(batchIDs, func(id string) any { return id })

	rows, err := q.QueryContext(ctx, `
		select id, batch_id, repo_did, name, knot_did, source_url, description, private, capability_token, status, attempts, next_attempt_at, error, created_at, updated_at
		from jobs where batch_id in (`+placeholders+`) order by batch_id asc, id asc
	`, args...)
	if err != nil {
		return nil, fmt.Errorf("querying jobs: %w", err)
	}
	defer rows.Close()

	jobs := make(map[string][]Job, len(batchIDs))
	for rows.Next() {
		var j Job
		var privInt int
		var errStr, capabilityToken sql.NullString
		var createdStr, updatedStr, nextAttStr string
		var statusStr string
		if err := rows.Scan(
			&j.ID,
			&j.BatchID,
			&j.RepoDid,
			&j.Name,
			&j.KnotDid,
			&j.SourceURL,
			&j.Description,
			&privInt,
			&capabilityToken,
			&statusStr,
			&j.Attempts,
			&nextAttStr,
			&errStr,
			&createdStr,
			&updatedStr,
		); err != nil {
			return nil, fmt.Errorf("scanning job: %w", err)
		}
		j.Private = privInt != 0
		if capabilityToken.Valid {
			j.CapabilityToken = &capabilityToken.String
		}
		j.Status = JobStatus(statusStr)
		if errStr.Valid {
			j.Error = &errStr.String
		}
		j.NextAttemptAt, _ = time.Parse(time.RFC3339, nextAttStr)
		j.CreatedAt, _ = time.Parse(time.RFC3339, createdStr)
		j.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr)
		jobs[j.BatchID] = append(jobs[j.BatchID], j)
	}
	return jobs, rows.Err()
}

func jobsForBatch(ctx context.Context, q querier, batchID string) ([]Job, error) {
	byBatch, err := jobsForBatches(ctx, q, []string{batchID})
	if err != nil {
		return nil, err
	}
	return byBatch[batchID], nil
}

func (d *DB) GetBatch(ctx context.Context, batchID string) (*Batch, []Job, error) {
	var b Batch
	var encToken sql.NullString
	var credExpStr sql.NullString
	var createdStr, updatedStr string

	err := d.QueryRowContext(ctx, `
		select id, owner_did, request_id, request_digest, encrypted_token, credential_expires_at, created_at, updated_at
		from batches where id = ?
	`, batchID).Scan(
		&b.ID,
		&b.OwnerDid,
		&b.RequestID,
		&b.RequestDigest,
		&encToken,
		&credExpStr,
		&createdStr,
		&updatedStr,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil, ErrNotFound
	}
	if err != nil {
		return nil, nil, fmt.Errorf("querying batch: %w", err)
	}

	if encToken.Valid {
		b.EncryptedToken = &encToken.String
	}
	if credExpStr.Valid {
		t, _ := time.Parse(time.RFC3339, credExpStr.String)
		b.CredentialExpiresAt = &t
	}
	b.CreatedAt, _ = time.Parse(time.RFC3339, createdStr)
	b.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr)

	jobs, err := jobsForBatch(ctx, d, batchID)
	if err != nil {
		return nil, nil, err
	}

	return &b, jobs, nil
}

func (d *DB) GetRecentTasksForOwner(ctx context.Context, ownerDid string, limit int) ([]Task, error) {
	if limit <= 0 {
		limit = 20
	}

	rows, err := d.QueryContext(ctx, `
		select id, owner_did, created_at, updated_at
		from batches
		where owner_did = ?
		order by created_at desc, id desc
		limit ?
	`, ownerDid, limit)
	if err != nil {
		return nil, fmt.Errorf("querying recent tasks: %w", err)
	}
	defer rows.Close()

	tasks := []Task{}
	for rows.Next() {
		var t Task
		var createdStr, updatedStr string
		if err := rows.Scan(&t.ID, &t.OwnerDid, &createdStr, &updatedStr); err != nil {
			return nil, fmt.Errorf("scanning task: %w", err)
		}
		t.CreatedAt, _ = time.Parse(time.RFC3339, createdStr)
		t.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr)
		tasks = append(tasks, t)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	ids := mapSlice(tasks, func(t Task) string { return t.ID })
	byBatch, err := jobsForBatches(ctx, d, ids)
	if err != nil {
		return nil, err
	}
	for i := range tasks {
		tasks[i].Jobs = byBatch[tasks[i].ID]
	}

	return tasks, nil
}

func (d *DB) GetJob(ctx context.Context, jobID int64) (*Job, *Batch, error) {
	var j Job
	var b Batch
	var privInt int
	var errStr sql.NullString
	var jCreatedStr, jUpdatedStr, nextStr string
	var bCreatedStr, bUpdatedStr string
	var encToken sql.NullString
	var capabilityToken sql.NullString
	var credExpStr sql.NullString
	var statusStr string

	err := d.QueryRowContext(ctx, `
		select j.id, j.batch_id, j.repo_did, j.name, j.knot_did, j.source_url, j.description, j.private, j.capability_token, j.status, j.attempts, j.next_attempt_at, j.error, j.created_at, j.updated_at,
		       b.id, b.owner_did, b.request_id, b.request_digest, b.encrypted_token, b.credential_expires_at, b.created_at, b.updated_at
		from jobs j
		join batches b on j.batch_id = b.id
		where j.id = ?
	`, jobID).Scan(
		&j.ID, &j.BatchID, &j.RepoDid, &j.Name, &j.KnotDid, &j.SourceURL, &j.Description, &privInt, &capabilityToken, &statusStr, &j.Attempts, &nextStr, &errStr, &jCreatedStr, &jUpdatedStr,
		&b.ID, &b.OwnerDid, &b.RequestID, &b.RequestDigest, &encToken, &credExpStr, &bCreatedStr, &bUpdatedStr,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil, ErrNotFound
	}
	if err != nil {
		return nil, nil, fmt.Errorf("querying job: %w", err)
	}

	j.Private = privInt != 0
	if capabilityToken.Valid {
		j.CapabilityToken = &capabilityToken.String
	}
	j.Status = JobStatus(statusStr)
	if errStr.Valid {
		j.Error = &errStr.String
	}
	j.NextAttemptAt, _ = time.Parse(time.RFC3339, nextStr)
	j.CreatedAt, _ = time.Parse(time.RFC3339, jCreatedStr)
	j.UpdatedAt, _ = time.Parse(time.RFC3339, jUpdatedStr)

	if encToken.Valid {
		b.EncryptedToken = &encToken.String
	}
	if credExpStr.Valid {
		t, _ := time.Parse(time.RFC3339, credExpStr.String)
		b.CredentialExpiresAt = &t
	}
	b.CreatedAt, _ = time.Parse(time.RFC3339, bCreatedStr)
	b.UpdatedAt, _ = time.Parse(time.RFC3339, bUpdatedStr)

	return &j, &b, nil
}

func (d *DB) ClaimNextQueuedJob(ctx context.Context) (*Job, *Batch, error) {
	tx, err := d.BeginTx(ctx, nil)
	if err != nil {
		return nil, nil, err
	}
	defer tx.Rollback()

	now := time.Now().UTC()
	nowStr := now.Format(time.RFC3339)

	// fair-share across owners so one queue cannot starve others
	query := `
		with owner_active as (
			select b.owner_did, count(*) as active
			from jobs j
			join batches b on j.batch_id = b.id
			where j.status in ('cloning', 'importing')
			group by b.owner_did
		)
		select j.id
		from jobs j
		join batches b on j.batch_id = b.id
		left join owner_active a on a.owner_did = b.owner_did
		where j.status = 'queued' and j.next_attempt_at <= ?
	`
	args := []any{nowStr}
	if d.MaxActivePerOwner > 0 {
		query += " and coalesce(a.active, 0) < ?"
		args = append(args, d.MaxActivePerOwner)
	}
	query += " order by coalesce(a.active, 0) asc, j.next_attempt_at asc, j.id asc limit 1"

	var jobID int64
	err = tx.QueryRowContext(ctx, query, args...).Scan(&jobID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil, nil
	}
	if err != nil {
		return nil, nil, fmt.Errorf("finding queued job: %w", err)
	}

	res, err := tx.ExecContext(ctx, `
		update jobs
		set status = 'cloning', attempts = attempts + 1, updated_at = ?
		where id = ? and status = 'queued'
	`, nowStr, jobID)
	if err != nil {
		return nil, nil, fmt.Errorf("claiming job: %w", err)
	}
	rowsAffected, _ := res.RowsAffected()
	if rowsAffected == 0 {
		return nil, nil, nil
	}

	if err := tx.Commit(); err != nil {
		return nil, nil, fmt.Errorf("committing claim: %w", err)
	}

	return d.GetJob(ctx, jobID)
}

func (d *DB) ReleaseJob(ctx context.Context, jobID int64, errMsg *string) error {
	nowStr := time.Now().UTC().Format(time.RFC3339)
	var errVal any
	if errMsg != nil {
		errVal = *errMsg
	}

	res, err := d.ExecContext(ctx, `
		update jobs
		set status = 'queued', attempts = max(attempts - 1, 0), next_attempt_at = ?, error = ?, updated_at = ?
		where id = ?
	`, nowStr, errVal, nowStr, jobID)
	if err != nil {
		return fmt.Errorf("releasing job: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return ErrNotFound
	}
	return nil
}

func (d *DB) UpdateJobStatus(ctx context.Context, jobID int64, status JobStatus, errMsg *string) error {
	nowStr := time.Now().UTC().Format(time.RFC3339)
	var errVal any
	if errMsg != nil {
		errVal = *errMsg
	}

	res, err := d.ExecContext(ctx, `
		update jobs
		set status = ?, error = ?, updated_at = ?
		where id = ?
	`, string(status), errVal, nowStr, jobID)
	if err != nil {
		return fmt.Errorf("updating job status: %w", err)
	}
	rowsAffected, _ := res.RowsAffected()
	if rowsAffected == 0 {
		return ErrNotFound
	}
	return nil
}

func (d *DB) ScheduleJobRetry(ctx context.Context, jobID int64, backoff time.Duration, errMsg *string) error {
	now := time.Now().UTC()
	nextAttemptAt := now.Add(backoff).Format(time.RFC3339)
	nowStr := now.Format(time.RFC3339)

	var errVal any
	if errMsg != nil {
		errVal = *errMsg
	}

	// repo_did, capability, and mirror survive retry so worker can resume
	res, err := d.ExecContext(ctx, `
		update jobs
		set status = 'queued', next_attempt_at = ?, error = ?, updated_at = ?
		where id = ?
	`, nextAttemptAt, errVal, nowStr, jobID)
	if err != nil {
		return fmt.Errorf("scheduling job retry: %w", err)
	}
	rowsAffected, _ := res.RowsAffected()
	if rowsAffected == 0 {
		return ErrNotFound
	}
	return nil
}

func (d *DB) RequeueJob(ctx context.Context, jobID int64, expectedOwnerDid string, newEncryptedToken *string) (*Batch, []Job, error) {
	tx, err := d.BeginTx(ctx, nil)
	if err != nil {
		return nil, nil, err
	}
	defer tx.Rollback()

	var batchID, ownerDid, currentStatus string
	var privateInt int
	err = tx.QueryRowContext(ctx, `
		select b.id, b.owner_did, j.status, j.private
		from jobs j
		join batches b on j.batch_id = b.id
		where j.id = ?
	`, jobID).Scan(&batchID, &ownerDid, &currentStatus, &privateInt)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil, ErrNotFound
	}
	if err != nil {
		return nil, nil, fmt.Errorf("finding job: %w", err)
	}

	if expectedOwnerDid != "" && ownerDid != expectedOwnerDid {
		return nil, nil, ErrNotFound
	}

	status := JobStatus(currentStatus)
	if !status.CanRetry() {
		return nil, nil, &InvalidTransitionError{Status: status}
	}

	now := time.Now().UTC()
	nowStr := now.Format(time.RFC3339)

	if newEncryptedToken != nil {
		newExpStr := now.Add(1 * time.Hour).Format(time.RFC3339)
		_, err = tx.ExecContext(ctx, `
			update batches
			set encrypted_token = ?, credential_expires_at = ?, updated_at = ?
			where id = ?
		`, *newEncryptedToken, newExpStr, nowStr, batchID)
		if err != nil {
			return nil, nil, fmt.Errorf("updating batch token: %w", err)
		}
	}

	// repo_did and capability survive manual retry so worker can resume
	_, err = tx.ExecContext(ctx, `
		update jobs
		set status = 'queued', attempts = 0, next_attempt_at = ?, error = null, updated_at = ?
		where id = ?
	`, nowStr, nowStr, jobID)
	if err != nil {
		return nil, nil, fmt.Errorf("requeueing job: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return nil, nil, fmt.Errorf("committing requeue: %w", err)
	}

	return d.GetBatch(ctx, batchID)
}

func (d *DB) SetJobCapability(ctx context.Context, jobID int64, token string) error {
	if token == "" {
		return errors.New("empty capability token")
	}
	res, err := d.ExecContext(ctx, `
		update jobs set capability_token = ?, updated_at = ? where id = ? and status = 'importing'
	`, token, time.Now().UTC().Format(time.RFC3339), jobID)
	if err != nil {
		return fmt.Errorf("publishing job capability: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return ErrNotFound
	}
	return nil
}

func (d *DB) RevokeJobCapability(ctx context.Context, jobID int64) error {
	_, err := d.ExecContext(ctx, `
		update jobs set capability_token = null, updated_at = ? where id = ?
	`, time.Now().UTC().Format(time.RFC3339), jobID)
	if err != nil {
		return fmt.Errorf("revoking job capability: %w", err)
	}
	return nil
}

func (d *DB) ResolveJobCapability(ctx context.Context, token, name string) (*Job, error) {
	var jobID int64
	// capability resolves on queued so a knot fetch can complete during retry backoff
	err := d.QueryRowContext(ctx, `
		select id from jobs
		where capability_token = ? and name = ?
		  and status in ('queued', 'cloning', 'importing')
	`, token, name).Scan(&jobID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("resolving job capability: %w", err)
	}
	job, _, err := d.GetJob(ctx, jobID)
	return job, err
}

func (d *DB) SetJobRepoDid(ctx context.Context, jobID int64, repoDid string) error {
	res, err := d.ExecContext(ctx, `
		update jobs set repo_did = ?, updated_at = ? where id = ?
	`, repoDid, time.Now().UTC().Format(time.RFC3339), jobID)
	if err != nil {
		return fmt.Errorf("recording created repo did: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return ErrNotFound
	}
	return nil
}

type EarlierRepoAttempt struct {
	RepoDid   string
	SourceURL string
}

func (d *DB) EarlierRepoAttemptsFor(ctx context.Context, ownerDid, knotDid, name string) ([]EarlierRepoAttempt, error) {
	rows, err := d.QueryContext(ctx, `
		select j.repo_did, j.source_url from jobs j
		join batches b on j.batch_id = b.id
		where b.owner_did = ? and j.knot_did = ? and j.name = ? and j.repo_did != ''
		order by j.id desc
	`, ownerDid, knotDid, name)
	if err != nil {
		return nil, fmt.Errorf("finding earlier repo attempts for %s/%s: %w", ownerDid, name, err)
	}
	defer rows.Close()

	var attempts []EarlierRepoAttempt
	for rows.Next() {
		var attempt EarlierRepoAttempt
		if err := rows.Scan(&attempt.RepoDid, &attempt.SourceURL); err != nil {
			return nil, fmt.Errorf("scanning an earlier repo attempt for %s/%s: %w", ownerDid, name, err)
		}
		attempts = append(attempts, attempt)
	}
	return attempts, rows.Err()
}

// Mirrors the request digest normalization.
func SameSource(a, b string) bool {
	normalize := func(raw string) string {
		return strings.TrimSuffix(strings.ToLower(strings.TrimSpace(raw)), ".git")
	}
	return normalize(a) != "" && normalize(a) == normalize(b)
}

func (d *DB) ResetInflightJobsToQueued(ctx context.Context) (int64, error) {
	nowStr := time.Now().UTC().Format(time.RFC3339)
	// repo_did survives crash recovery so the worker can resume from the knot
	res, err := d.ExecContext(ctx, `
		update jobs
		set status = 'queued', next_attempt_at = ?, capability_token = null, updated_at = ?
		where status in ('cloning', 'importing')
	`, nowStr, nowStr)
	if err != nil {
		return 0, fmt.Errorf("resetting inflight jobs: %w", err)
	}
	return res.RowsAffected()
}

func (d *DB) CheckAndScrubBatchToken(ctx context.Context, batchID string) (bool, error) {
	tx, err := d.BeginTx(ctx, nil)
	if err != nil {
		return false, err
	}
	defer tx.Rollback()

	var nonTerminalCount int
	err = tx.QueryRowContext(ctx, `
		select count(*) from jobs
		where batch_id = ? and status not in ('completed', 'failed', 'authorization_required')
	`, batchID).Scan(&nonTerminalCount)
	if err != nil {
		return false, fmt.Errorf("counting non-terminal jobs: %w", err)
	}

	if nonTerminalCount == 0 {
		nowStr := time.Now().UTC().Format(time.RFC3339)
		_, err = tx.ExecContext(ctx, `
			update batches
			set encrypted_token = null, updated_at = ?
			where id = ? and encrypted_token is not null
		`, nowStr, batchID)
		if err != nil {
			return false, fmt.Errorf("scrubbing batch token: %w", err)
		}
		if err := tx.Commit(); err != nil {
			return false, fmt.Errorf("committing scrub: %w", err)
		}
		return true, nil
	}

	return false, nil
}

func (d *DB) ScrubExpiredCredentials(ctx context.Context) (int64, error) {
	tx, err := d.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()

	now := time.Now().UTC()
	nowStr := now.Format(time.RFC3339)

	errMsg := "GitHub credential expired after 1 hour grant window"
	_, err = tx.ExecContext(ctx, `
		update jobs
		set status = 'authorization_required', error = ?, updated_at = ?
		where private = 1 and status = 'queued' and batch_id in (
			select id from batches where encrypted_token is not null and credential_expires_at <= ?
		)
	`, errMsg, nowStr, nowStr)
	if err != nil {
		return 0, fmt.Errorf("marking expired unstarted jobs: %w", err)
	}

	res, err := tx.ExecContext(ctx, `
		update batches
		set encrypted_token = null, updated_at = ?
		where encrypted_token is not null and credential_expires_at <= ?
	`, nowStr, nowStr)
	if err != nil {
		return 0, fmt.Errorf("scrubbing expired batch tokens: %w", err)
	}

	affected, _ := res.RowsAffected()

	if err := tx.Commit(); err != nil {
		return 0, fmt.Errorf("committing credential expiration: %w", err)
	}

	return affected, nil
}