popiart-server/internal/server/sqlite_repo.go

package server

import (
	"crypto/aes"
	"crypto/cipher"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"

	_ "modernc.org/sqlite"
)

type sqliteRepository struct {
	db        *sql.DB
	secretKey [32]byte
}

func newSQLiteRepository(dbPath, sessionSecret string) (*sqliteRepository, error) {
	if strings.TrimSpace(dbPath) == "" {
		return nil, errors.New("sqlite path is required")
	}
	if err := os.MkdirAll(filepath.Dir(dbPath), 0o755); err != nil {
		return nil, fmt.Errorf("create sqlite directory: %w", err)
	}

	db, err := sql.Open("sqlite", dbPath)
	if err != nil {
		return nil, fmt.Errorf("open sqlite: %w", err)
	}
	db.SetMaxOpenConns(1)

	repo := &sqliteRepository{
		db:        db,
		secretKey: sha256.Sum256([]byte(defaultString(strings.TrimSpace(sessionSecret), "popiart-dev-session-secret"))),
	}
	if err := repo.migrate(); err != nil {
		_ = db.Close()
		return nil, err
	}
	return repo, nil
}

func (r *sqliteRepository) migrate() error {
	statements := []string{
		`PRAGMA foreign_keys = ON;`,
		`CREATE TABLE IF NOT EXISTS sessions (
			session_id TEXT PRIMARY KEY,
			user_id TEXT NOT NULL,
			user_json TEXT NOT NULL,
			token_enc BLOB NOT NULL,
			token_masked TEXT NOT NULL,
			gateway_user_id INTEGER,
			gateway_access_token_enc BLOB,
			created_at TEXT NOT NULL,
			expires_at TEXT NOT NULL
		);`,
		`CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id);`,
		`CREATE TABLE IF NOT EXISTS jobs (
			job_id TEXT PRIMARY KEY,
			user_id TEXT NOT NULL,
			session_id TEXT,
			project_id TEXT,
			skill_id TEXT,
			route_key TEXT,
			model_id TEXT,
			exec_mode TEXT NOT NULL,
			newapi_task_id TEXT,
			status TEXT NOT NULL,
			input_json TEXT NOT NULL,
			result_refs_json TEXT,
			usage_json TEXT,
			error_json TEXT,
			idempotency_key TEXT,
			created_at TEXT NOT NULL,
			started_at TEXT,
			finished_at TEXT,
			UNIQUE(user_id, idempotency_key)
		);`,
		`CREATE INDEX IF NOT EXISTS idx_jobs_user_created_at ON jobs(user_id, created_at DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_jobs_status_created_at ON jobs(status, created_at DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_jobs_project_created_at ON jobs(project_id, created_at DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_jobs_skill_created_at ON jobs(skill_id, created_at DESC);`,
		`CREATE TABLE IF NOT EXISTS media_records (
			media_id TEXT PRIMARY KEY,
			user_id TEXT NOT NULL,
			artifact_id TEXT,
			project_id TEXT,
			filename TEXT NOT NULL,
			content_type TEXT NOT NULL,
			size_bytes INTEGER NOT NULL,
			created_at TEXT NOT NULL,
			url TEXT NOT NULL,
			visibility TEXT,
			sha256 TEXT,
			local_path TEXT NOT NULL
		);`,
		`CREATE INDEX IF NOT EXISTS idx_media_records_user_created_at ON media_records(user_id, created_at DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_media_records_project_created_at ON media_records(project_id, created_at DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_media_records_artifact_id ON media_records(artifact_id);`,
		`CREATE TABLE IF NOT EXISTS artifacts (
			artifact_id TEXT PRIMARY KEY,
			user_id TEXT NOT NULL,
			job_id TEXT NOT NULL,
			project_id TEXT,
			result_index INTEGER NOT NULL,
			media_id TEXT,
			filename TEXT NOT NULL,
			content_type TEXT NOT NULL,
			size_bytes INTEGER NOT NULL,
			created_at TEXT NOT NULL,
			expires_at TEXT,
			visibility TEXT,
			sha256 TEXT,
			storage_status TEXT,
			source_skill_id TEXT,
			source_model_id TEXT,
			source_route_key TEXT,
			source_input_json TEXT,
			prompt_text TEXT,
			usage_json TEXT,
			ref_json TEXT NOT NULL
		);`,
		`CREATE INDEX IF NOT EXISTS idx_artifacts_user_created_at ON artifacts(user_id, created_at DESC, artifact_id DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_artifacts_job_result_index ON artifacts(job_id, result_index ASC);`,
		`CREATE INDEX IF NOT EXISTS idx_artifacts_project_created_at ON artifacts(project_id, created_at DESC, artifact_id DESC);`,
		`CREATE INDEX IF NOT EXISTS idx_artifacts_media_id ON artifacts(media_id);`,
		`CREATE TABLE IF NOT EXISTS skill_routes (
			route_key TEXT NOT NULL,
			scope_key TEXT NOT NULL,
			model_id TEXT NOT NULL,
			options_json TEXT,
			updated_at TEXT NOT NULL,
			PRIMARY KEY(route_key, scope_key)
		);`,
	}
	for _, stmt := range statements {
		if _, err := r.db.Exec(stmt); err != nil {
			return fmt.Errorf("apply sqlite migration: %w", err)
		}
	}
	for _, stmt := range []string{
		`ALTER TABLE jobs ADD COLUMN session_id TEXT;`,
		`ALTER TABLE jobs ADD COLUMN route_key TEXT;`,
		`ALTER TABLE jobs ADD COLUMN exec_mode TEXT NOT NULL DEFAULT 'sync_result';`,
		`ALTER TABLE jobs ADD COLUMN newapi_task_id TEXT;`,
		`ALTER TABLE jobs ADD COLUMN result_refs_json TEXT;`,
	} {
		if _, err := r.db.Exec(stmt); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column name") {
			return fmt.Errorf("upgrade sqlite jobs schema: %w", err)
		}
	}
	for _, stmt := range []string{
		`ALTER TABLE sessions ADD COLUMN gateway_user_id INTEGER;`,
		`ALTER TABLE sessions ADD COLUMN gateway_access_token_enc BLOB;`,
	} {
		if _, err := r.db.Exec(stmt); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column name") {
			return fmt.Errorf("upgrade sqlite sessions schema: %w", err)
		}
	}
	return nil
}

func (r *sqliteRepository) CreateSession(upstreamKey string, u user, ttl time.Duration) (session, error) {
	now := time.Now().UTC()
	current := session{
		Token:       "sess_" + randomID(),
		UserID:      u.ID,
		UpstreamKey: upstreamKey,
		User:        u,
		CreatedAt:   now,
		ExpiresAt:   now.Add(ttl),
	}

	enc, err := r.encryptSecret(upstreamKey)
	if err != nil {
		return session{}, err
	}
	userJSON, err := marshalJSON(u)
	if err != nil {
		return session{}, err
	}
	if _, err := r.db.Exec(
		`INSERT INTO sessions (session_id, user_id, user_json, token_enc, token_masked, created_at, expires_at)
		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
		current.Token,
		current.UserID,
		userJSON,
		enc,
		maskSecret(upstreamKey),
		now.Format(time.RFC3339),
		current.ExpiresAt.Format(time.RFC3339),
	); err != nil {
		return session{}, fmt.Errorf("insert session: %w", err)
	}
	return current, nil
}

func (r *sqliteRepository) GetSession(token string) (session, bool, error) {
	row := r.db.QueryRow(
		`SELECT user_id, user_json, token_enc, gateway_user_id, gateway_access_token_enc, created_at, expires_at
		 FROM sessions WHERE session_id = ?`,
		token,
	)
	var (
		current               session
		userJSON              string
		tokenEnc              []byte
		gatewayUserID         sql.NullInt64
		gatewayAccessTokenEnc []byte
		createdAt, expiresAt  string
	)
	if err := row.Scan(
		&current.UserID,
		&userJSON,
		&tokenEnc,
		&gatewayUserID,
		&gatewayAccessTokenEnc,
		&createdAt,
		&expiresAt,
	); errors.Is(err, sql.ErrNoRows) {
		return session{}, false, nil
	} else if err != nil {
		return session{}, false, fmt.Errorf("scan session: %w", err)
	}

	if err := unmarshalJSON(userJSON, &current.User); err != nil {
		return session{}, false, fmt.Errorf("decode session user: %w", err)
	}
	upstreamKey, err := r.decryptSecret(tokenEnc)
	if err != nil {
		return session{}, false, err
	}
	if gatewayUserID.Valid {
		current.GatewayUserID = int(gatewayUserID.Int64)
	}
	if len(gatewayAccessTokenEnc) > 0 {
		gatewayAccessToken, err := r.decryptSecret(gatewayAccessTokenEnc)
		if err != nil {
			return session{}, false, err
		}
		current.GatewayAccessToken = gatewayAccessToken
	}
	current.Token = token
	current.UpstreamKey = upstreamKey
	current.CreatedAt = parseRFC3339(createdAt)
	current.ExpiresAt = parseRFC3339(expiresAt)
	if !current.ExpiresAt.IsZero() && time.Now().UTC().After(current.ExpiresAt) {
		_ = r.DeleteSession(token)
		return session{}, false, nil
	}
	return current, true, nil
}

func (r *sqliteRepository) DeleteSession(token string) error {
	_, err := r.db.Exec(`DELETE FROM sessions WHERE session_id = ?`, token)
	if err != nil {
		return fmt.Errorf("delete session: %w", err)
	}
	return nil
}

func (r *sqliteRepository) RotateSession(oldToken string, ttl time.Duration) (session, bool, error) {
	current, exists, err := r.GetSession(oldToken)
	if err != nil || !exists {
		return session{}, exists, err
	}
	if err := r.DeleteSession(oldToken); err != nil {
		return session{}, false, err
	}
	next, err := r.CreateSession(current.UpstreamKey, current.User, ttl)
	if err != nil {
		return session{}, false, err
	}
	if current.GatewayUserID > 0 && strings.TrimSpace(current.GatewayAccessToken) != "" {
		if err := r.UpdateSessionGatewayBinding(next.Token, current.GatewayUserID, current.GatewayAccessToken); err != nil {
			return session{}, false, err
		}
		next.GatewayUserID = current.GatewayUserID
		next.GatewayAccessToken = current.GatewayAccessToken
	}
	return next, true, nil
}

func (r *sqliteRepository) UpdateSessionGatewayBinding(token string, gatewayUserID int, gatewayAccessToken string) error {
	if strings.TrimSpace(token) == "" {
		return errors.New("session token is required")
	}
	if gatewayUserID <= 0 {
		return errors.New("gateway user id is required")
	}
	if strings.TrimSpace(gatewayAccessToken) == "" {
		return errors.New("gateway access token is required")
	}
	enc, err := r.encryptSecret(gatewayAccessToken)
	if err != nil {
		return err
	}
	if _, err := r.db.Exec(
		`UPDATE sessions SET gateway_user_id = ?, gateway_access_token_enc = ? WHERE session_id = ?`,
		gatewayUserID,
		enc,
		token,
	); err != nil {
		return fmt.Errorf("update session gateway binding: %w", err)
	}
	return nil
}

func (r *sqliteRepository) CreateJob(record *job) (*job, int, error) {
	if record == nil {
		return nil, 0, errors.New("job record is required")
	}
	if record.UserID == "" {
		return nil, 0, errors.New("user_id is required")
	}
	if record.JobID == "" {
		return nil, 0, errors.New("job_id is required")
	}
	if record.CreatedAt == "" {
		record.CreatedAt = time.Now().UTC().Format(time.RFC3339)
	}
	if record.Status == "" {
		record.Status = "pending"
	}

	if record.IdempotencyKey != "" {
		existingID, ok, err := r.findIdempotentJob(record.UserID, record.IdempotencyKey)
		if err != nil {
			return nil, 0, err
		}
		if ok {
			existing, exists, err := r.GetJob(existingID)
			if err != nil {
				return nil, 0, err
			}
			if exists {
				return existing, httpStatusOK, nil
			}
		}
	}

	inputJSON, err := marshalJSON(record.Input)
	if err != nil {
		return nil, 0, err
	}

	if _, err := r.db.Exec(
		`INSERT INTO jobs (
			job_id, user_id, session_id, project_id, skill_id, route_key, model_id,
			exec_mode, newapi_task_id, status, input_json, result_refs_json, usage_json,
			error_json, idempotency_key, created_at, started_at, finished_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, NULL, NULL)`,
		record.JobID,
		record.UserID,
		nullableString(record.SessionID),
		nullableString(record.ProjectID),
		nullableString(record.SkillID),
		nullableString(record.RouteKey),
		nullableString(record.ModelID),
		defaultString(record.ExecMode, "sync_result"),
		nullableString(record.NewAPITaskID),
		record.Status,
		inputJSON,
		nullableString(record.IdempotencyKey),
		record.CreatedAt,
	); err != nil {
		return nil, 0, fmt.Errorf("insert job: %w", err)
	}
	return cloneJob(record), httpStatusAccepted, nil
}

func (r *sqliteRepository) GetJob(jobID string) (*job, bool, error) {
	row := r.db.QueryRow(
		`SELECT job_id, user_id, session_id, project_id, skill_id, route_key, model_id,
		        exec_mode, newapi_task_id, status, input_json, result_refs_json, usage_json,
		        error_json, idempotency_key, created_at, started_at, finished_at
		   FROM jobs WHERE job_id = ?`,
		jobID,
	)
	return scanJob(row)
}

func (r *sqliteRepository) ListJobs(userID, status, skillID, projectID string, limit, offset int) ([]job, int, error) {
	where, args := buildJobFilters(userID, status, skillID, projectID)
	var total int
	if err := r.db.QueryRow(`SELECT COUNT(*) FROM jobs`+where, args...).Scan(&total); err != nil {
		return nil, 0, fmt.Errorf("count jobs: %w", err)
	}

	argsWithPaging := append(append([]any{}, args...), limit, offset)
	rows, err := r.db.Query(
		`SELECT job_id, user_id, session_id, project_id, skill_id, route_key, model_id,
		        exec_mode, newapi_task_id, status, input_json, result_refs_json, usage_json,
		        error_json, idempotency_key, created_at, started_at, finished_at
		   FROM jobs`+where+` ORDER BY datetime(created_at) DESC, job_id DESC LIMIT ? OFFSET ?`,
		argsWithPaging...,
	)
	if err != nil {
		return nil, 0, fmt.Errorf("list jobs: %w", err)
	}
	defer rows.Close()

	items := make([]job, 0, limit)
	for rows.Next() {
		record, _, err := scanJob(rows)
		if err != nil {
			return nil, 0, err
		}
		items = append(items, *record)
	}
	if err := rows.Err(); err != nil {
		return nil, 0, fmt.Errorf("iterate jobs: %w", err)
	}
	return items, total, nil
}

func (r *sqliteRepository) MarkJobRunning(jobID string) (*job, bool, error) {
	record, exists, err := r.GetJob(jobID)
	if err != nil || !exists {
		return nil, exists, err
	}
	if isTerminalStatus(record.Status) {
		return record, true, nil
	}
	now := time.Now().UTC().Format(time.RFC3339)
	if _, err := r.db.Exec(
		`UPDATE jobs SET status = ?, started_at = COALESCE(started_at, ?) WHERE job_id = ?`,
		"running", now, jobID,
	); err != nil {
		return nil, false, fmt.Errorf("mark running job: %w", err)
	}
	return r.GetJob(jobID)
}

func (r *sqliteRepository) LinkUpstreamTask(jobID, upstreamTaskID string) error {
	record, exists, err := r.GetJob(jobID)
	if err != nil {
		return err
	}
	if !exists || isTerminalStatus(record.Status) {
		return nil
	}
	now := time.Now().UTC().Format(time.RFC3339)
	if _, err := r.db.Exec(
		`UPDATE jobs
		    SET status = ?, started_at = COALESCE(started_at, ?), newapi_task_id = ?
		  WHERE job_id = ?`,
		"running", now, nullableString(upstreamTaskID), jobID,
	); err != nil {
		return fmt.Errorf("link upstream task: %w", err)
	}
	return nil
}

func (r *sqliteRepository) FailJob(jobID, code, message string, details map[string]any) error {
	record, exists, err := r.GetJob(jobID)
	if err != nil {
		return err
	}
	if !exists || isTerminalStatus(record.Status) {
		return nil
	}
	now := time.Now().UTC().Format(time.RFC3339)
	errorJSON, err := marshalJSON(&jobError{Code: code, Message: message, Details: details})
	if err != nil {
		return err
	}
	if _, err := r.db.Exec(
		`UPDATE jobs SET status = ?, error_json = ?, finished_at = ? WHERE job_id = ?`,
		"failed", errorJSON, now, jobID,
	); err != nil {
		return fmt.Errorf("update failed job: %w", err)
	}
	return nil
}

func (r *sqliteRepository) CompleteSyncResult(jobID string, refs []resultRef, usage map[string]any) error {
	record, exists, err := r.GetJob(jobID)
	if err != nil {
		return err
	}
	if !exists || isTerminalStatus(record.Status) {
		return nil
	}
	now := time.Now().UTC().Format(time.RFC3339)
	resultJSON, err := marshalJSON(refs)
	if err != nil {
		return err
	}
	usageJSON, err := marshalNullableJSON(usage)
	if err != nil {
		return err
	}
	if _, err := r.db.Exec(
		`UPDATE jobs
		    SET status = ?, result_refs_json = ?, usage_json = ?, finished_at = ?
		  WHERE job_id = ?`,
		"done", resultJSON, usageJSON, now, jobID,
	); err != nil {
		return fmt.Errorf("complete sync result job: %w", err)
	}
	return nil
}

func (r *sqliteRepository) CancelJob(userID, jobID string) (*job, bool, bool, error) {
	record, exists, err := r.GetJob(jobID)
	if err != nil || !exists {
		return nil, exists, false, err
	}
	if record.UserID != userID {
		return nil, false, false, nil
	}
	if isTerminalStatus(record.Status) {
		return nil, true, true, nil
	}
	now := time.Now().UTC().Format(time.RFC3339)
	if _, err := r.db.Exec(
		`UPDATE jobs SET status = ?, finished_at = ? WHERE job_id = ?`,
		"cancelled", now, jobID,
	); err != nil {
		return nil, false, false, fmt.Errorf("cancel job: %w", err)
	}
	record, exists, err = r.GetJob(jobID)
	return record, exists, false, err
}

func (r *sqliteRepository) GetRoutes(projectID string) (map[string]string, error) {
	rows, err := r.db.Query(
		`SELECT route_key, model_id FROM skill_routes WHERE scope_key = ? ORDER BY route_key ASC`,
		scopeKey(projectID),
	)
	if err != nil {
		return nil, fmt.Errorf("query project routes: %w", err)
	}
	defer rows.Close()

	items := map[string]string{}
	for rows.Next() {
		var routeKey, modelID string
		if err := rows.Scan(&routeKey, &modelID); err != nil {
			return nil, fmt.Errorf("scan project route: %w", err)
		}
		items[routeKey] = modelID
	}
	return items, rows.Err()
}

func (r *sqliteRepository) SetRoute(projectID, routeKey, modelID string) error {
	_, err := r.db.Exec(
		`INSERT INTO skill_routes (route_key, scope_key, model_id, options_json, updated_at)
		 VALUES (?, ?, ?, NULL, ?)
		 ON CONFLICT(route_key, scope_key)
		 DO UPDATE SET model_id = excluded.model_id, updated_at = excluded.updated_at`,
		routeKey,
		scopeKey(projectID),
		modelID,
		time.Now().UTC().Format(time.RFC3339),
	)
	if err != nil {
		return fmt.Errorf("upsert route override: %w", err)
	}
	return nil
}

func (r *sqliteRepository) UnsetRoute(projectID, routeKey string) error {
	if _, err := r.db.Exec(
		`DELETE FROM skill_routes WHERE route_key = ? AND scope_key = ?`,
		routeKey,
		scopeKey(projectID),
	); err != nil {
		return fmt.Errorf("delete route override: %w", err)
	}
	return nil
}

func (r *sqliteRepository) UpsertMedia(record mediaRecord) error {
	_, err := r.db.Exec(
		`INSERT INTO media_records (
			media_id, user_id, artifact_id, project_id, filename, content_type,
			size_bytes, created_at, url, visibility, sha256, local_path
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(media_id) DO UPDATE SET
			user_id = excluded.user_id,
			artifact_id = excluded.artifact_id,
			project_id = excluded.project_id,
			filename = excluded.filename,
			content_type = excluded.content_type,
			size_bytes = excluded.size_bytes,
			created_at = excluded.created_at,
			url = excluded.url,
			visibility = excluded.visibility,
			sha256 = excluded.sha256,
			local_path = excluded.local_path`,
		record.ID,
		record.UserID,
		nullableString(record.ArtifactID),
		nullableString(record.ProjectID),
		record.Filename,
		record.ContentType,
		record.SizeBytes,
		record.CreatedAt,
		record.URL,
		nullableString(record.Visibility),
		nullableString(record.SHA256),
		record.LocalPath,
	)
	if err != nil {
		return fmt.Errorf("upsert media: %w", err)
	}
	return nil
}

func (r *sqliteRepository) GetMedia(mediaID string) (*mediaRecord, bool, error) {
	row := r.db.QueryRow(
		`SELECT media_id, user_id, artifact_id, project_id, filename, content_type,
		        size_bytes, created_at, url, visibility, sha256, local_path
		   FROM media_records WHERE media_id = ?`,
		mediaID,
	)
	var (
		record                mediaRecord
		artifactID, projectID sql.NullString
		visibility, sha256    sql.NullString
	)
	err := row.Scan(
		&record.ID,
		&record.UserID,
		&artifactID,
		&projectID,
		&record.Filename,
		&record.ContentType,
		&record.SizeBytes,
		&record.CreatedAt,
		&record.URL,
		&visibility,
		&sha256,
		&record.LocalPath,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, fmt.Errorf("scan media: %w", err)
	}
	record.ArtifactID = artifactID.String
	record.ProjectID = projectID.String
	record.Visibility = visibility.String
	record.SHA256 = sha256.String
	return &record, true, nil
}

func (r *sqliteRepository) UpsertArtifacts(userID, projectID string, items []artifact) error {
	if strings.TrimSpace(userID) == "" || len(items) == 0 {
		return nil
	}
	tx, err := r.db.Begin()
	if err != nil {
		return fmt.Errorf("begin artifacts upsert: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	stmt, err := tx.Prepare(
		`INSERT INTO artifacts (
			artifact_id, user_id, job_id, project_id, result_index, media_id,
			filename, content_type, size_bytes, created_at, expires_at,
			visibility, sha256, storage_status, source_skill_id, source_model_id,
			source_route_key, source_input_json, prompt_text, usage_json, ref_json
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(artifact_id) DO UPDATE SET
			user_id = excluded.user_id,
			job_id = excluded.job_id,
			project_id = excluded.project_id,
			result_index = excluded.result_index,
			media_id = excluded.media_id,
			filename = excluded.filename,
			content_type = excluded.content_type,
			size_bytes = excluded.size_bytes,
			created_at = excluded.created_at,
			expires_at = excluded.expires_at,
			visibility = excluded.visibility,
			sha256 = excluded.sha256,
			storage_status = excluded.storage_status,
			source_skill_id = excluded.source_skill_id,
			source_model_id = excluded.source_model_id,
			source_route_key = excluded.source_route_key,
			source_input_json = excluded.source_input_json,
			prompt_text = excluded.prompt_text,
			usage_json = excluded.usage_json,
			ref_json = excluded.ref_json`,
	)
	if err != nil {
		return fmt.Errorf("prepare artifacts upsert: %w", err)
	}
	defer stmt.Close()

	for _, item := range items {
		_, idx, ok := parseArtifactID(item.ID)
		if !ok {
			return fmt.Errorf("invalid artifact id: %s", item.ID)
		}
		refJSON, err := marshalJSON(item.Ref)
		if err != nil {
			return err
		}
		sourceInputJSON, err := marshalNullableJSON(item.SourceInput)
		if err != nil {
			return err
		}
		usageJSON, err := marshalNullableJSON(item.Usage)
		if err != nil {
			return err
		}
		if _, err := stmt.Exec(
			item.ID,
			userID,
			item.JobID,
			nullableString(defaultString(item.ProjectID, projectID)),
			idx,
			nullableString(item.MediaID),
			item.Filename,
			item.ContentType,
			item.SizeBytes,
			item.CreatedAt,
			nullableString(item.ExpiresAt),
			nullableString(item.Visibility),
			nullableString(item.SHA256),
			nullableString(item.StorageStatus),
			nullableString(item.SourceSkillID),
			nullableString(item.SourceModelID),
			nullableString(item.SourceRouteKey),
			sourceInputJSON,
			nullableString(item.PromptText),
			usageJSON,
			refJSON,
		); err != nil {
			return fmt.Errorf("exec artifacts upsert: %w", err)
		}
	}

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

func (r *sqliteRepository) GetArtifact(userID, artifactID string) (*artifact, resultRef, bool, error) {
	row := r.db.QueryRow(
		`SELECT artifact_id, job_id, project_id, media_id, filename, content_type, size_bytes,
		        created_at, expires_at, visibility, sha256, storage_status, source_skill_id,
		        source_model_id, source_route_key, source_input_json, prompt_text, usage_json, ref_json
		   FROM artifacts WHERE user_id = ? AND artifact_id = ?`,
		userID, artifactID,
	)
	return scanArtifact(row)
}

func (r *sqliteRepository) ListArtifacts(userID, projectID, jobID string, limit, offset int) ([]artifact, int, error) {
	if strings.TrimSpace(userID) == "" {
		return nil, 0, errors.New("user_id is required")
	}
	where, args := buildArtifactFilters(userID, projectID, jobID)
	var total int
	if err := r.db.QueryRow(`SELECT COUNT(*) FROM artifacts`+where, args...).Scan(&total); err != nil {
		return nil, 0, fmt.Errorf("count artifacts: %w", err)
	}

	argsWithPaging := append(append([]any{}, args...), limit, offset)
	rows, err := r.db.Query(
		`SELECT artifact_id, job_id, project_id, media_id, filename, content_type, size_bytes,
		        created_at, expires_at, visibility, sha256, storage_status, source_skill_id,
		        source_model_id, source_route_key, source_input_json, prompt_text, usage_json, ref_json
		   FROM artifacts`+where+` ORDER BY datetime(created_at) DESC, artifact_id DESC LIMIT ? OFFSET ?`,
		argsWithPaging...,
	)
	if err != nil {
		return nil, 0, fmt.Errorf("list artifacts: %w", err)
	}
	defer rows.Close()

	items := make([]artifact, 0, limit)
	for rows.Next() {
		item, _, exists, err := scanArtifact(rows)
		if err != nil {
			return nil, 0, err
		}
		if exists && item != nil {
			items = append(items, *item)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, 0, fmt.Errorf("iterate artifacts: %w", err)
	}
	return items, total, nil
}

func (r *sqliteRepository) findIdempotentJob(userID, idem string) (string, bool, error) {
	var jobID string
	err := r.db.QueryRow(
		`SELECT job_id FROM jobs WHERE user_id = ? AND idempotency_key = ?`,
		userID, idem,
	).Scan(&jobID)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, fmt.Errorf("lookup idempotent job: %w", err)
	}
	return jobID, true, nil
}

func (r *sqliteRepository) encryptSecret(secret string) ([]byte, error) {
	block, err := aes.NewCipher(r.secretKey[:])
	if err != nil {
		return nil, fmt.Errorf("build session cipher: %w", err)
	}
	gcm, err := cipher.NewGCM(block)
	if err != nil {
		return nil, fmt.Errorf("build session gcm: %w", err)
	}
	nonce := make([]byte, gcm.NonceSize())
	if _, err := rand.Read(nonce); err != nil {
		return nil, fmt.Errorf("generate session nonce: %w", err)
	}
	return gcm.Seal(nonce, nonce, []byte(secret), nil), nil
}

func (r *sqliteRepository) decryptSecret(payload []byte) (string, error) {
	block, err := aes.NewCipher(r.secretKey[:])
	if err != nil {
		return "", fmt.Errorf("build session cipher: %w", err)
	}
	gcm, err := cipher.NewGCM(block)
	if err != nil {
		return "", fmt.Errorf("build session gcm: %w", err)
	}
	if len(payload) < gcm.NonceSize() {
		return "", errors.New("invalid encrypted session token")
	}
	nonce := payload[:gcm.NonceSize()]
	plaintext, err := gcm.Open(nil, nonce, payload[gcm.NonceSize():], nil)
	if err != nil {
		return "", fmt.Errorf("decrypt session token: %w", err)
	}
	return string(plaintext), nil
}

func buildJobFilters(userID, status, skillID, projectID string) (string, []any) {
	filters := []string{"user_id = ?"}
	args := []any{userID}
	if status != "" {
		filters = append(filters, "status = ?")
		args = append(args, status)
	}
	if skillID != "" {
		filters = append(filters, "skill_id = ?")
		args = append(args, skillID)
	}
	if projectID != "" {
		filters = append(filters, "project_id = ?")
		args = append(args, projectID)
	}
	return " WHERE " + strings.Join(filters, " AND "), args
}

func buildArtifactFilters(userID, projectID, jobID string) (string, []any) {
	filters := []string{"user_id = ?"}
	args := []any{userID}
	if projectID != "" {
		filters = append(filters, "project_id = ?")
		args = append(args, projectID)
	}
	if jobID != "" {
		filters = append(filters, "job_id = ?")
		args = append(args, jobID)
	}
	return " WHERE " + strings.Join(filters, " AND "), args
}

func scanArtifact(row interface{ Scan(dest ...any) error }) (*artifact, resultRef, bool, error) {
	var (
		item                                         artifact
		projectID, mediaID, expiresAt                sql.NullString
		visibility, sha256, storageStatus            sql.NullString
		sourceSkillID, sourceModelID, sourceRouteKey sql.NullString
		sourceInputJSON, promptText, usageJSON       sql.NullString
		refJSON                                      string
	)
	err := row.Scan(
		&item.ID,
		&item.JobID,
		&projectID,
		&mediaID,
		&item.Filename,
		&item.ContentType,
		&item.SizeBytes,
		&item.CreatedAt,
		&expiresAt,
		&visibility,
		&sha256,
		&storageStatus,
		&sourceSkillID,
		&sourceModelID,
		&sourceRouteKey,
		&sourceInputJSON,
		&promptText,
		&usageJSON,
		&refJSON,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, resultRef{}, false, nil
	}
	if err != nil {
		return nil, resultRef{}, false, fmt.Errorf("scan artifact: %w", err)
	}

	item.ProjectID = projectID.String
	item.MediaID = mediaID.String
	item.ExpiresAt = expiresAt.String
	item.Visibility = visibility.String
	item.SHA256 = sha256.String
	item.StorageStatus = storageStatus.String
	item.SourceSkillID = sourceSkillID.String
	item.SourceModelID = sourceModelID.String
	item.SourceRouteKey = sourceRouteKey.String
	item.PromptText = promptText.String
	if sourceInputJSON.Valid {
		if err := unmarshalJSON(sourceInputJSON.String, &item.SourceInput); err != nil {
			return nil, resultRef{}, false, fmt.Errorf("decode artifact source input: %w", err)
		}
	}
	if usageJSON.Valid {
		if err := unmarshalJSON(usageJSON.String, &item.Usage); err != nil {
			return nil, resultRef{}, false, fmt.Errorf("decode artifact usage: %w", err)
		}
	}

	var ref resultRef
	if err := unmarshalJSON(refJSON, &ref); err != nil {
		return nil, resultRef{}, false, fmt.Errorf("decode artifact ref: %w", err)
	}
	item.Ref = ref
	item.URL = strings.TrimSpace(ref.URL)
	return &item, ref, true, nil
}

func scanJob(row interface{ Scan(dest ...any) error }) (*job, bool, error) {
	var (
		record                                                             job
		sessionID, projectID, skillID, routeKey, modelID                   sql.NullString
		execMode, newapiTaskID, resultRefsJSON, usageJSON, errorJSON, idem sql.NullString
		startedAt, finishedAt                                              sql.NullString
		inputJSON                                                          string
	)
	err := row.Scan(
		&record.JobID,
		&record.UserID,
		&sessionID,
		&projectID,
		&skillID,
		&routeKey,
		&modelID,
		&execMode,
		&newapiTaskID,
		&record.Status,
		&inputJSON,
		&resultRefsJSON,
		&usageJSON,
		&errorJSON,
		&idem,
		&record.CreatedAt,
		&startedAt,
		&finishedAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, fmt.Errorf("scan job: %w", err)
	}

	record.SessionID = sessionID.String
	record.ProjectID = projectID.String
	record.SkillID = skillID.String
	record.RouteKey = routeKey.String
	record.ModelID = modelID.String
	record.ExecMode = execMode.String
	record.NewAPITaskID = newapiTaskID.String
	record.IdempotencyKey = idem.String
	record.StartedAt = startedAt.String
	record.FinishedAt = finishedAt.String

	if err := unmarshalJSON(inputJSON, &record.Input); err != nil {
		return nil, false, fmt.Errorf("decode job input: %w", err)
	}
	if resultRefsJSON.Valid {
		if err := unmarshalJSON(resultRefsJSON.String, &record.ResultRefs); err != nil {
			return nil, false, fmt.Errorf("decode job result refs: %w", err)
		}
	}
	if usageJSON.Valid {
		if err := unmarshalJSON(usageJSON.String, &record.Usage); err != nil {
			return nil, false, fmt.Errorf("decode job usage: %w", err)
		}
	}
	if errorJSON.Valid {
		var decoded jobError
		if err := unmarshalJSON(errorJSON.String, &decoded); err != nil {
			return nil, false, fmt.Errorf("decode job error: %w", err)
		}
		record.Error = &decoded
	}
	record.ArtifactIDs = buildArtifactIDs(record.JobID, record.ResultRefs)
	return &record, true, nil
}

func marshalJSON(value any) (string, error) {
	data, err := json.Marshal(value)
	if err != nil {
		return "", fmt.Errorf("marshal json: %w", err)
	}
	return string(data), nil
}

func marshalNullableJSON(value any) (any, error) {
	if value == nil {
		return nil, nil
	}
	return marshalJSON(value)
}

func unmarshalJSON(value string, dst any) error {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return json.Unmarshal([]byte(value), dst)
}

func nullableString(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return value
}

func parseRFC3339(value string) time.Time {
	if strings.TrimSpace(value) == "" {
		return time.Time{}
	}
	ts, err := time.Parse(time.RFC3339, value)
	if err != nil {
		return time.Time{}
	}
	return ts
}

func isTerminalStatus(status string) bool {
	switch status {
	case "done", "failed", "cancelled":
		return true
	default:
		return false
	}
}

func scopeKey(projectID string) string {
	projectID = strings.TrimSpace(projectID)
	if projectID == "" {
		return "__global__"
	}
	return "project:" + projectID
}

const (
	httpStatusOK       = 200
	httpStatusAccepted = 202
)