From 2edc941fa9bd264436d4932aef6c68e4542d56b9 Mon Sep 17 00:00:00 2001 From: groot Date: Mon, 10 Aug 2026 06:43:00 +0200 Subject: [PATCH] RC-3-C --- .dockerignore | 2 +- .gitignore | 2 +- internal/data/profile_cleanup_test.go | 72 ++ internal/data/schema.sql | 130 ++ internal/data/store.go | 1602 +++++++++++++++++++++++++ internal/data/successor_test.go | 68 ++ 6 files changed, 1874 insertions(+), 2 deletions(-) create mode 100644 internal/data/profile_cleanup_test.go create mode 100644 internal/data/schema.sql create mode 100644 internal/data/store.go create mode 100644 internal/data/successor_test.go diff --git a/.dockerignore b/.dockerignore index 6119bb3..6c9cce7 100644 --- a/.dockerignore +++ b/.dockerignore @@ -2,7 +2,7 @@ .env web/node_modules web/dist -data +/data *.db *.db-shm *.db-wal diff --git a/.gitignore b/.gitignore index 0b0ca15..6aa3723 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,7 @@ .env web/node_modules web/dist -data/ +/data/ *.db *.db-shm *.db-wal diff --git a/internal/data/profile_cleanup_test.go b/internal/data/profile_cleanup_test.go new file mode 100644 index 0000000..77ad6fc --- /dev/null +++ b/internal/data/profile_cleanup_test.go @@ -0,0 +1,72 @@ +package data + +import ( + "context" + "testing" + "time" + + "neuralhunt/internal/auth" +) + +func TestInactiveNonWinnerCleanupProtectsWinnersAndRecentClients(t *testing.T) { + ctx := context.Background() + db, err := OpenSQLite(ctx, t.TempDir()+"/cleanup.db") + if err != nil { + t.Fatal(err) + } + defer db.Close() + s := New(db) + jwk := auth.PublicJWK{Kty: "EC", Crv: "P-256", X: "AQ", Y: "Ag"} + for _, id := range []string{"old_delete", "old_winner", "recent_keep"} { + if err := s.UpsertClient(ctx, id, jwk); err != nil { + t.Fatal(err) + } + } + old := time.Now().UTC().Add(-45 * 24 * time.Hour).UnixMilli() + recent := time.Now().UTC().Add(-2 * time.Hour).UnixMilli() + if _, err := db.ExecContext(ctx, `UPDATE clients SET last_seen=? WHERE id IN ('old_delete','old_winner')`, old); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `UPDATE clients SET last_seen=? WHERE id='recent_keep'`, recent); err != nil { + t.Fatal(err) + } + if err := s.EnsureActiveTasks(ctx, 1, 32); err != nil { + t.Fatal(err) + } + var taskID string + if err := db.QueryRowContext(ctx, `SELECT id FROM tasks LIMIT 1`).Scan(&taskID); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, `UPDATE tasks SET winner_client_id=? WHERE id=?`, "old_winner", taskID); err != nil { + t.Fatal(err) + } + + cutoff := time.Now().UTC().Add(-30 * 24 * time.Hour).UnixMilli() + candidates, err := s.InactiveNonWinnerClients(ctx, cutoff) + if err != nil { + t.Fatal(err) + } + if len(candidates) != 1 || candidates[0].ClientID != "old_delete" { + t.Fatalf("unexpected candidates: %+v", candidates) + } + wins, err := s.OldWinnerCount(ctx, cutoff) + if err != nil || wins != 1 { + t.Fatalf("protected winner count=%d err=%v", wins, err) + } + deleted, err := s.DeleteInactiveNonWinnerClients(ctx, cutoff, []string{"old_delete", "old_winner", "recent_keep"}) + if err != nil { + t.Fatal(err) + } + if len(deleted) != 1 || deleted[0] != "old_delete" { + t.Fatalf("unexpected deleted: %#v", deleted) + } + if s.ClientExists(ctx, "old_delete") { + t.Fatal("old non-winner should have been deleted") + } + if !s.ClientExists(ctx, "old_winner") { + t.Fatal("winner must be protected") + } + if !s.ClientExists(ctx, "recent_keep") { + t.Fatal("recent client must be protected") + } +} diff --git a/internal/data/schema.sql b/internal/data/schema.sql new file mode 100644 index 0000000..eb5ae1c --- /dev/null +++ b/internal/data/schema.sql @@ -0,0 +1,130 @@ +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS clients ( + id TEXT PRIMARY KEY, + public_jwk TEXT NOT NULL, + created_at INTEGER NOT NULL, + last_seen INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS tasks ( + id TEXT PRIMARY KEY, + secret TEXT NOT NULL, + public_seed TEXT NOT NULL, + range_bits INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active','completed','closed')), + paused INTEGER NOT NULL DEFAULT 0, + guess_min_interval_sec INTEGER, + client_submit_interval_sec INTEGER, + revision INTEGER NOT NULL DEFAULT 0, + parent_task_id TEXT REFERENCES tasks(id), + display_name TEXT NOT NULL DEFAULT '', + description TEXT NOT NULL DEFAULT '', + nft_prompt_instructions TEXT NOT NULL DEFAULT '', + nft_negative_prompt TEXT NOT NULL DEFAULT '', + nft_style_reference TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL, + completed_at INTEGER, + winner_client_id TEXT REFERENCES clients(id), + winner_signature TEXT, + winning_guess TEXT, + artifact_status TEXT NOT NULL DEFAULT 'none' CHECK (artifact_status IN ('none','pending','generating','ready','error')), + artifact_uri TEXT, + artifact_manifest_uri TEXT, + artifact_error TEXT +); +CREATE INDEX IF NOT EXISTS tasks_status_created_idx ON tasks(status, created_at); + +-- False guesses are intentionally not persisted. This table only keeps the +-- aggregate state needed for the 3D map, ranking, rate limiting and next seq. +CREATE TABLE IF NOT EXISTS task_points ( + task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + client_id TEXT NOT NULL REFERENCES clients(id) ON DELETE CASCADE, + score REAL NOT NULL DEFAULT 0, + x REAL NOT NULL DEFAULT 0, + y REAL NOT NULL DEFAULT 0, + z REAL NOT NULL DEFAULT 0, + guess_count INTEGER NOT NULL DEFAULT 0, + next_seq INTEGER NOT NULL DEFAULT 0, + last_guess_at INTEGER, + PRIMARY KEY (task_id, client_id) +); +CREATE INDEX IF NOT EXISTS task_points_rank_idx ON task_points(task_id, score DESC); + + + +-- A browser or shell identity can explicitly choose which active task it works +-- on. The selection is persisted so switching devices with the same exported +-- identity keeps the chosen task (while single-active-connection enforcement +-- still applies at runtime). +CREATE TABLE IF NOT EXISTS client_task_selection ( + client_id TEXT PRIMARY KEY REFERENCES clients(id) ON DELETE CASCADE, + task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + updated_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS client_task_selection_task_idx ON client_task_selection(task_id); + +CREATE TABLE IF NOT EXISTS client_unlocks ( + client_id TEXT NOT NULL REFERENCES clients(id) ON DELETE CASCADE, + unlock_key TEXT NOT NULL, + task_id TEXT REFERENCES tasks(id) ON DELETE SET NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (client_id, unlock_key) +); + +-- A short lease enforces one active websocket per browser identity. +CREATE TABLE IF NOT EXISTS presence_leases ( + client_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + expires_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS presence_exp_idx ON presence_leases(expires_at); + +-- Admin task actions may run immediately or at a future timestamp. Payloads are +-- small JSON objects validated by the Go server before being scheduled. +CREATE TABLE IF NOT EXISTS task_actions ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + action_type TEXT NOT NULL, + payload_json TEXT NOT NULL DEFAULT '{}', + execute_at INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','running','done','error','cancelled')), + created_at INTEGER NOT NULL, + executed_at INTEGER, + error TEXT +); +CREATE INDEX IF NOT EXISTS task_actions_due_idx ON task_actions(status, execute_at); +CREATE INDEX IF NOT EXISTS task_actions_task_idx ON task_actions(task_id, created_at DESC); + +-- Successful external image-model calls are logged with the token usage +-- returned by the provider. Costs are local estimates based on pinned public +-- standard token rates. Billing reconciliation still belongs to the provider. +CREATE TABLE IF NOT EXISTS artifact_api_usage ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at INTEGER NOT NULL, + task_id TEXT REFERENCES tasks(id) ON DELETE SET NULL, + kind TEXT NOT NULL CHECK (kind IN ('character_anchor','artifact')), + provider TEXT NOT NULL, + model TEXT NOT NULL, + endpoint TEXT NOT NULL, + size TEXT NOT NULL, + quality TEXT NOT NULL, + request_id TEXT, + input_tokens INTEGER NOT NULL DEFAULT 0, + input_text_tokens INTEGER NOT NULL DEFAULT 0, + input_image_tokens INTEGER NOT NULL DEFAULT 0, + output_tokens INTEGER NOT NULL DEFAULT 0, + total_tokens INTEGER NOT NULL DEFAULT 0, + estimated_cost_usd REAL, + pricing_basis TEXT NOT NULL DEFAULT '', + meta_json TEXT NOT NULL DEFAULT '{}' +); +CREATE INDEX IF NOT EXISTS artifact_api_usage_created_idx ON artifact_api_usage(created_at DESC); +CREATE INDEX IF NOT EXISTS artifact_api_usage_kind_created_idx ON artifact_api_usage(kind, created_at DESC); +CREATE INDEX IF NOT EXISTS artifact_api_usage_task_idx ON artifact_api_usage(task_id); diff --git a/internal/data/store.go b/internal/data/store.go new file mode 100644 index 0000000..657083f --- /dev/null +++ b/internal/data/store.go @@ -0,0 +1,1602 @@ +package data + +import ( + "context" + "crypto/rand" + "database/sql" + _ "embed" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "math" + "math/big" + "net/url" + "os" + "path/filepath" + "strings" + "time" + + "neuralhunt/internal/auth" + "neuralhunt/internal/core" + + _ "modernc.org/sqlite" +) + +//go:embed schema.sql +var schemaSQL string + +var ( + ErrRateLimited = errors.New("guess rate limited") + ErrBadSequence = errors.New("unexpected guess sequence") + ErrTaskCompleted = errors.New("task already completed") + ErrTaskPaused = errors.New("task paused") + ErrPresenceBusy = errors.New("identity already connected") +) + +type Store struct{ DB *sql.DB } + +func New(db *sql.DB) *Store { return &Store{DB: db} } + +// OpenSQLite opens an embedded SQLite database using modernc.org/sqlite. +// WAL is intentionally enabled only for local-disk/same-host deployments. +func OpenSQLite(ctx context.Context, path string) (*sql.DB, error) { + if path == "" { + path = "./data/neuralhunt.db" + } + abs, err := filepath.Abs(path) + if err != nil { + return nil, err + } + if err := os.MkdirAll(filepath.Dir(abs), 0o750); err != nil { + return nil, err + } + slashPath := filepath.ToSlash(abs) + if vol := filepath.VolumeName(abs); vol != "" && !strings.HasPrefix(slashPath, "/") { + slashPath = "/" + slashPath + } + u := &url.URL{Scheme: "file", Path: slashPath} + q := u.Query() + q.Add("_pragma", "busy_timeout(10000)") + q.Add("_pragma", "foreign_keys(ON)") + q.Add("_pragma", "synchronous(NORMAL)") + q.Set("_txlock", "immediate") + u.RawQuery = q.Encode() + + db, err := sql.Open("sqlite", u.String()) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(4) + db.SetMaxIdleConns(4) + db.SetConnMaxLifetime(0) + if err := db.PingContext(ctx); err != nil { + db.Close() + return nil, fmt.Errorf("open sqlite database %q: %w", abs, err) + } + conn, err := db.Conn(ctx) + if err != nil { + db.Close() + return nil, fmt.Errorf("sqlite connection for WAL %q: %w", abs, err) + } + var journalMode string + if err := conn.QueryRowContext(ctx, `PRAGMA journal_mode=WAL`).Scan(&journalMode); err != nil { + _ = conn.Close() + db.Close() + return nil, fmt.Errorf("enable sqlite WAL %q: %w", abs, err) + } + _ = conn.Close() + for i, stmt := range strings.Split(schemaSQL, ";") { + stmt = strings.TrimSpace(stmt) + if stmt == "" { + continue + } + if _, err := db.ExecContext(ctx, stmt); err != nil { + db.Close() + return nil, fmt.Errorf("sqlite migrate statement %d: %w", i+1, err) + } + } + // CREATE TABLE IF NOT EXISTS does not add columns to databases from older + // Neural Hunt releases, so evolve the tasks table explicitly. + for _, m := range []struct{ name, def string }{ + {"paused", "INTEGER NOT NULL DEFAULT 0"}, + {"guess_min_interval_sec", "INTEGER"}, + {"client_submit_interval_sec", "INTEGER"}, + {"revision", "INTEGER NOT NULL DEFAULT 0"}, + {"parent_task_id", "TEXT REFERENCES tasks(id)"}, + {"display_name", "TEXT NOT NULL DEFAULT ''"}, + {"description", "TEXT NOT NULL DEFAULT ''"}, + {"nft_prompt_instructions", "TEXT NOT NULL DEFAULT ''"}, + {"nft_negative_prompt", "TEXT NOT NULL DEFAULT ''"}, + {"nft_style_reference", "TEXT NOT NULL DEFAULT ''"}, + } { + if err := ensureColumn(ctx, db, "tasks", m.name, m.def); err != nil { + db.Close() + return nil, fmt.Errorf("sqlite add tasks.%s: %w", m.name, err) + } + } + // This index must be created after legacy databases have received the new + // parent_task_id column; keeping it in schema.sql would make upgrades from + // V2.4 fail before ensureColumn gets a chance to run. + if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS tasks_parent_unique_idx ON tasks(parent_task_id) WHERE parent_task_id IS NOT NULL`); err != nil { + db.Close() + return nil, fmt.Errorf("sqlite create successor index: %w", err) + } + return db, nil +} + +func ensureColumn(ctx context.Context, db *sql.DB, table, name, def string) error { + rows, err := db.QueryContext(ctx, `PRAGMA table_info(`+table+`)`) + if err != nil { + return err + } + found := false + for rows.Next() { + var cid int + var col, typ string + var notnull, pk int + var dflt sql.NullString + if err := rows.Scan(&cid, &col, &typ, ¬null, &dflt, &pk); err != nil { + rows.Close() + return err + } + if strings.EqualFold(col, name) { + found = true + } + } + if err := rows.Close(); err != nil { + return err + } + if found { + return nil + } + _, err = db.ExecContext(ctx, `ALTER TABLE `+table+` ADD COLUMN `+name+` `+def) + return err +} + +func NewID(prefix string) string { + b := make([]byte, 16) + _, _ = rand.Read(b) + return prefix + hex.EncodeToString(b) +} + +func fromUnixMS(v int64) time.Time { + if v <= 0 { + return time.Time{} + } + return time.UnixMilli(v).UTC() +} + +func (s *Store) UpsertClient(ctx context.Context, id string, jwk auth.PublicJWK) error { + b, _ := json.Marshal(jwk) + now := time.Now().UTC().UnixMilli() + _, err := s.DB.ExecContext(ctx, `INSERT INTO clients(id,public_jwk,created_at,last_seen) VALUES(?,?,?,?) + ON CONFLICT(id) DO UPDATE SET last_seen=excluded.last_seen`, id, string(b), now, now) + return err +} + +func (s *Store) ClientExists(ctx context.Context, id string) bool { + var n int + if err := s.DB.QueryRowContext(ctx, `SELECT count(*) FROM clients WHERE id=?`, id).Scan(&n); err != nil { + return false + } + return n == 1 +} + +func (s *Store) ClientPublicJWK(ctx context.Context, id string) (auth.PublicJWK, error) { + var raw string + if err := s.DB.QueryRowContext(ctx, `SELECT public_jwk FROM clients WHERE id=?`, id).Scan(&raw); err != nil { + return auth.PublicJWK{}, err + } + var jwk auth.PublicJWK + if err := json.Unmarshal([]byte(raw), &jwk); err != nil { + return auth.PublicJWK{}, err + } + return jwk, nil +} + +type Task struct { + ID, PublicSeed string + RangeBits int + Status string + Paused bool + GuessMinIntervalSec *int + ClientSubmitIntervalSec *int + Revision int64 + ParentTaskID *string + DisplayName string + Description string + NFTPromptInstructions string + NFTNegativePrompt string + NFTStyleReference string + CreatedAt time.Time + CompletedAt *time.Time + WinnerClientID *string + ArtifactStatus string + ArtifactURI *string + ArtifactManifestURI *string +} + +const taskColumns = `id,public_seed,range_bits,status,paused,guess_min_interval_sec,client_submit_interval_sec,revision,parent_task_id,display_name,description,nft_prompt_instructions,nft_negative_prompt,nft_style_reference,created_at,completed_at,winner_client_id,artifact_status,artifact_uri,artifact_manifest_uri` + +func scanTask(scanner interface{ Scan(...any) error }, withSecret bool) (Task, string, error) { + var t Task + var created int64 + var completed sql.NullInt64 + var winner, artifactURI, manifestURI, parent sql.NullString + var guessMin, clientSubmit sql.NullInt64 + var paused int + var secret string + args := []any{&t.ID, &t.PublicSeed, &t.RangeBits, &t.Status, &paused, &guessMin, &clientSubmit, &t.Revision, &parent, &t.DisplayName, &t.Description, &t.NFTPromptInstructions, &t.NFTNegativePrompt, &t.NFTStyleReference, &created, &completed, &winner, &t.ArtifactStatus, &artifactURI, &manifestURI} + if withSecret { + args = append(args, &secret) + } + if err := scanner.Scan(args...); err != nil { + return Task{}, "", err + } + t.Paused = paused != 0 + if parent.Valid { + v := parent.String + t.ParentTaskID = &v + } + if guessMin.Valid { + v := int(guessMin.Int64) + t.GuessMinIntervalSec = &v + } + if clientSubmit.Valid { + v := int(clientSubmit.Int64) + t.ClientSubmitIntervalSec = &v + } + t.CreatedAt = fromUnixMS(created) + if completed.Valid { + v := fromUnixMS(completed.Int64) + t.CompletedAt = &v + } + if winner.Valid { + v := winner.String + t.WinnerClientID = &v + } + if artifactURI.Valid { + v := artifactURI.String + t.ArtifactURI = &v + } + if manifestURI.Valid { + v := manifestURI.String + t.ArtifactManifestURI = &v + } + return t, secret, nil +} + +func (s *Store) ActiveTasks(ctx context.Context) ([]Task, error) { + rows, err := s.DB.QueryContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status='active' ORDER BY created_at,id`) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]Task, 0) + for rows.Next() { + t, _, err := scanTask(rows, false) + if err != nil { + return nil, err + } + out = append(out, t) + } + return out, rows.Err() +} + +func (s *Store) insertTaskTx(ctx context.Context, tx *sql.Tx, bits int, parent *Task) (Task, error) { + if bits < 8 || bits > 128 { + return Task{}, fmt.Errorf("bits must be 8..128") + } + secret, err := core.RandomDecimal(bits) + if err != nil { + return Task{}, err + } + seed, err := core.RandomSeed() + if err != nil { + return Task{}, err + } + id := NewID("task_") + now := time.Now().UTC().UnixMilli() + var parentID any + var guessMin, clientSubmit any + name, description, promptInstructions, negativePrompt, styleReference := "", "", "", "", "" + if parent != nil { + parentID = parent.ID + if parent.GuessMinIntervalSec != nil { + guessMin = *parent.GuessMinIntervalSec + } + if parent.ClientSubmitIntervalSec != nil { + clientSubmit = *parent.ClientSubmitIntervalSec + } + name = parent.DisplayName + description = parent.Description + promptInstructions = parent.NFTPromptInstructions + negativePrompt = parent.NFTNegativePrompt + styleReference = parent.NFTStyleReference + } + _, err = tx.ExecContext(ctx, `INSERT INTO tasks( + id,secret,public_seed,range_bits,status,paused,guess_min_interval_sec,client_submit_interval_sec,revision, + parent_task_id,display_name,description,nft_prompt_instructions,nft_negative_prompt,nft_style_reference,created_at,artifact_status) + VALUES(?,?,?,?,'active',0,?,?,0,?,?,?,?,?,?,?,'none')`, + id, secret, seed, bits, guessMin, clientSubmit, parentID, name, description, promptInstructions, negativePrompt, styleReference, now) + if err != nil { + return Task{}, err + } + row := tx.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE id=?`, id) + t, _, err := scanTask(row, false) + return t, err +} + +// EnsureSuccessorTask creates exactly one follow-up task for a completed/closed +// predecessor. The successor inherits the predecessor's live task configuration, +// presentation metadata and per-task NFT prompt instructions. Existing client +// selections are moved atomically to the successor so browser and shell clients +// continue the same task series after a win/close. +func (s *Store) EnsureSuccessorTask(ctx context.Context, predecessorID string, fallbackBits int) (Task, error) { + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return Task{}, err + } + defer tx.Rollback() + if row := tx.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE parent_task_id=? LIMIT 1`, predecessorID); row != nil { + if t, _, scanErr := scanTask(row, false); scanErr == nil { + return t, tx.Commit() + } else if !errors.Is(scanErr, sql.ErrNoRows) { + return Task{}, scanErr + } + } + row := tx.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE id=?`, predecessorID) + pred, _, err := scanTask(row, false) + if err != nil { + return Task{}, err + } + if pred.Status == "active" { + return Task{}, fmt.Errorf("predecessor is still active") + } + bits := pred.RangeBits + if bits < 8 || bits > 128 { + bits = fallbackBits + } + next, err := s.insertTaskTx(ctx, tx, bits, &pred) + if err != nil { + // A concurrent creator may have won the unique parent_task_id race. + if row := tx.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE parent_task_id=? LIMIT 1`, predecessorID); row != nil { + if existing, _, scanErr := scanTask(row, false); scanErr == nil { + return existing, tx.Commit() + } + } + return Task{}, err + } + if _, err := tx.ExecContext(ctx, `UPDATE client_task_selection SET task_id=?,updated_at=? WHERE task_id=?`, next.ID, time.Now().UTC().UnixMilli(), predecessorID); err != nil { + return Task{}, err + } + if err := tx.Commit(); err != nil { + return Task{}, err + } + return next, nil +} + +func (s *Store) EnsureActiveTasks(ctx context.Context, count, bits int) error { + if count < 1 { + count = 1 + } + for { + var n int + if err := s.DB.QueryRowContext(ctx, `SELECT count(*) FROM tasks WHERE status='active'`).Scan(&n); err != nil { + return err + } + if n >= count { + return nil + } + // Prefer replacing a finished task with its inherited successor before + // creating a generic task from global defaults. + var predID string + err := s.DB.QueryRowContext(ctx, `SELECT t.id FROM tasks t + WHERE t.status IN ('completed','closed') AND NOT EXISTS(SELECT 1 FROM tasks c WHERE c.parent_task_id=t.id) + ORDER BY COALESCE(t.completed_at,t.created_at) DESC,t.id DESC LIMIT 1`).Scan(&predID) + if err == nil { + if _, err := s.EnsureSuccessorTask(ctx, predID, bits); err != nil { + return err + } + continue + } + if !errors.Is(err, sql.ErrNoRows) { + return err + } + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return err + } + if _, err := s.insertTaskTx(ctx, tx, bits, nil); err != nil { + _ = tx.Rollback() + return err + } + if err := tx.Commit(); err != nil { + return err + } + } +} + +func chooseIndex(cid string, n int) int { + if n <= 1 { + return 0 + } + var x uint64 + for i := 0; i < len(cid); i++ { + x = x*131 + uint64(cid[i]) + } + return int(x % uint64(n)) +} + +func (s *Store) SetClientTaskSelection(ctx context.Context, cid, taskID string) error { + var status string + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM tasks WHERE id=?`, taskID).Scan(&status); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("task not found") + } + return err + } + if status != "active" { + return fmt.Errorf("task is not active") + } + _, err := s.DB.ExecContext(ctx, `INSERT INTO client_task_selection(client_id,task_id,updated_at) VALUES(?,?,?) + ON CONFLICT(client_id) DO UPDATE SET task_id=excluded.task_id,updated_at=excluded.updated_at`, cid, taskID, time.Now().UTC().UnixMilli()) + return err +} + +func (s *Store) SelectedTaskID(ctx context.Context, cid string) (string, error) { + var id string + err := s.DB.QueryRowContext(ctx, `SELECT s.task_id FROM client_task_selection s JOIN tasks t ON t.id=s.task_id WHERE s.client_id=? AND t.status='active'`, cid).Scan(&id) + return id, err +} + +func (s *Store) TaskForClient(ctx context.Context, cid string) (Task, error) { + if id, err := s.SelectedTaskID(ctx, cid); err == nil { + row := s.DB.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE id=? AND status='active'`, id) + t, _, err := scanTask(row, false) + if err == nil { + return t, nil + } + } else if !errors.Is(err, sql.ErrNoRows) { + return Task{}, err + } + ts, err := s.ActiveTasks(ctx) + if err != nil { + return Task{}, err + } + if len(ts) == 0 { + return Task{}, sql.ErrNoRows + } + t := ts[chooseIndex(cid, len(ts))] + if err := s.SetClientTaskSelection(ctx, cid, t.ID); err != nil { + return Task{}, err + } + return t, nil +} + +type ClientTask struct { + ID string `json:"id"` + DisplayName string `json:"display_name"` + Description string `json:"description"` + RangeBits int `json:"range_bits"` + Paused bool `json:"paused"` + Revision int64 `json:"revision"` + CreatedAt time.Time `json:"created_at"` + PointCount int `json:"point_count"` + OwnScore float64 `json:"own_score"` + OwnRank int64 `json:"own_rank"` + Selected bool `json:"selected"` + GuessMinIntervalSec *int `json:"guess_min_interval_sec,omitempty"` + ClientSubmitIntervalSec *int `json:"client_submit_interval_sec,omitempty"` + HasCustomStyleReference bool `json:"has_custom_style_reference"` +} + +func (s *Store) ActiveTasksForClient(ctx context.Context, cid string) ([]ClientTask, error) { + selected, _ := s.SelectedTaskID(ctx, cid) + rows, err := s.DB.QueryContext(ctx, `SELECT t.id,t.display_name,t.description,t.range_bits,t.paused,t.revision,t.created_at,t.guess_min_interval_sec,t.client_submit_interval_sec, + CASE WHEN trim(COALESCE(t.nft_style_reference,''))<>'' THEN 1 ELSE 0 END, + (SELECT count(*) FROM task_points p WHERE p.task_id=t.id), + COALESCE((SELECT p.score FROM task_points p WHERE p.task_id=t.id AND p.client_id=?),0), + CASE WHEN EXISTS(SELECT 1 FROM task_points p0 WHERE p0.task_id=t.id AND p0.client_id=?) THEN + (SELECT 1+count(*) FROM task_points p2 WHERE p2.task_id=t.id AND p2.score>(SELECT p3.score FROM task_points p3 WHERE p3.task_id=t.id AND p3.client_id=?)) + ELSE 0 END + FROM tasks t WHERE t.status='active' ORDER BY t.created_at,t.id`, cid, cid, cid) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]ClientTask, 0) + for rows.Next() { + var t ClientTask + var paused int + var created int64 + var guessMin, clientSubmit sql.NullInt64 + var hasCustomStyle int + if err := rows.Scan(&t.ID, &t.DisplayName, &t.Description, &t.RangeBits, &paused, &t.Revision, &created, &guessMin, &clientSubmit, &hasCustomStyle, &t.PointCount, &t.OwnScore, &t.OwnRank); err != nil { + return nil, err + } + t.Paused = paused != 0 + t.CreatedAt = fromUnixMS(created) + t.Selected = t.ID == selected + t.HasCustomStyleReference = hasCustomStyle != 0 + if guessMin.Valid { + v := int(guessMin.Int64) + t.GuessMinIntervalSec = &v + } + if clientSubmit.Valid { + v := int(clientSubmit.Int64) + t.ClientSubmitIntervalSec = &v + } + out = append(out, t) + } + return out, rows.Err() +} + +func (s *Store) NextSeq(ctx context.Context, taskID, cid string) (int64, error) { + var seq int64 + err := s.DB.QueryRowContext(ctx, `SELECT next_seq FROM task_points WHERE task_id=? AND client_id=?`, taskID, cid).Scan(&seq) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + return seq, err +} + +type SecretTask struct { + Task + Secret string +} + +func (s *Store) SecretTask(ctx context.Context, id string) (SecretTask, error) { + row := s.DB.QueryRowContext(ctx, `SELECT `+taskColumns+`,secret FROM tasks WHERE id=?`, id) + t, secret, err := scanTask(row, true) + if err != nil { + return SecretTask{}, err + } + return SecretTask{Task: t, Secret: secret}, nil +} + +type Point struct { + ClientID string `json:"client_id"` + Score float64 `json:"score"` + X float64 `json:"x"` + Y float64 `json:"y"` + Z float64 `json:"z"` + GuessCount int64 `json:"guess_count"` + Rank int64 `json:"rank"` + LastGuessAt time.Time `json:"last_guess_at"` +} + +func (s *Store) EnsurePoint(ctx context.Context, taskID, cid string) (Point, error) { + x, y, z := core.Position(cid, 0) + if _, err := s.DB.ExecContext(ctx, `INSERT INTO task_points(task_id,client_id,score,x,y,z,guess_count,next_seq,last_guess_at) + VALUES(?,?,?,?,?,?,0,0,NULL) ON CONFLICT(task_id,client_id) DO NOTHING`, taskID, cid, 0, x, y, z); err != nil { + return Point{}, err + } + var p Point + var last sql.NullInt64 + if err := s.DB.QueryRowContext(ctx, `SELECT client_id,score,x,y,z,guess_count,last_guess_at, + 1+(SELECT count(*) FROM task_points p2 WHERE p2.task_id=p.task_id AND p2.score>p.score) + FROM task_points p WHERE task_id=? AND client_id=?`, taskID, cid). + Scan(&p.ClientID, &p.Score, &p.X, &p.Y, &p.Z, &p.GuessCount, &last, &p.Rank); err != nil { + return Point{}, err + } + if last.Valid { + p.LastGuessAt = fromUnixMS(last.Int64) + } + return p, nil +} + +func (s *Store) SubmitGuess(ctx context.Context, t SecretTask, cid string, seq int64, guess, sig string, score float64, correct bool, minInterval time.Duration) (Point, error) { + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return Point{}, err + } + defer tx.Rollback() + + var status string + var paused int + if err := tx.QueryRowContext(ctx, `SELECT status,paused FROM tasks WHERE id=?`, t.ID).Scan(&status, &paused); err != nil { + return Point{}, err + } + if status != "active" { + return Point{}, ErrTaskCompleted + } + if paused != 0 { + return Point{}, ErrTaskPaused + } + + var oldScore float64 + var nextSeq int64 + var lastMS sql.NullInt64 + err = tx.QueryRowContext(ctx, `SELECT score,next_seq,last_guess_at FROM task_points WHERE task_id=? AND client_id=?`, t.ID, cid).Scan(&oldScore, &nextSeq, &lastMS) + exists := err == nil + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return Point{}, err + } + if !exists { + nextSeq = 0 + } + if seq != nextSeq { + return Point{}, ErrBadSequence + } + now := time.Now().UTC() + nowMS := now.UnixMilli() + if lastMS.Valid && minInterval > 0 && nowMS-lastMS.Int64 < minInterval.Milliseconds() { + return Point{}, ErrRateLimited + } + + best := score + if exists && oldScore > best { + best = oldScore + } + x, y, z := core.Position(cid, best) + if exists { + _, err = tx.ExecContext(ctx, `UPDATE task_points SET score=?,x=?,y=?,z=?,guess_count=guess_count+1,next_seq=next_seq+1,last_guess_at=? WHERE task_id=? AND client_id=?`, best, x, y, z, nowMS, t.ID, cid) + } else { + _, err = tx.ExecContext(ctx, `INSERT INTO task_points(task_id,client_id,score,x,y,z,guess_count,next_seq,last_guess_at) VALUES(?,?,?,?,?,?,1,1,?)`, t.ID, cid, best, x, y, z, nowMS) + } + if err != nil { + return Point{}, err + } + + if correct { + res, err := tx.ExecContext(ctx, `UPDATE tasks SET status='completed',completed_at=?,winner_client_id=?,winner_signature=?,winning_guess=?,artifact_status='pending',revision=revision+1 WHERE id=? AND status='active'`, nowMS, cid, sig, guess, t.ID) + if err != nil { + return Point{}, err + } + n, _ := res.RowsAffected() + if n != 1 { + return Point{}, ErrTaskCompleted + } + if err := s.unlocksTx(ctx, tx, cid, t.ID, nowMS); err != nil { + return Point{}, err + } + } + + var p Point + var last int64 + if err = tx.QueryRowContext(ctx, `SELECT client_id,score,x,y,z,guess_count,last_guess_at FROM task_points WHERE task_id=? AND client_id=?`, t.ID, cid).Scan(&p.ClientID, &p.Score, &p.X, &p.Y, &p.Z, &p.GuessCount, &last); err != nil { + return Point{}, err + } + p.LastGuessAt = fromUnixMS(last) + if err = tx.QueryRowContext(ctx, `SELECT 1+count(*) FROM task_points WHERE task_id=? AND score>?`, t.ID, p.Score).Scan(&p.Rank); err != nil { + return Point{}, err + } + if err = tx.Commit(); err != nil { + return Point{}, err + } + return p, nil +} + +func (s *Store) unlocksTx(ctx context.Context, tx *sql.Tx, cid, taskID string, nowMS int64) error { + var wins int + if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM tasks WHERE winner_client_id=? AND status='completed'`, cid).Scan(&wins); err != nil { + return err + } + keys := []struct { + n int + k string + }{{1, "first_win"}, {3, "three_wins"}, {10, "ten_wins"}} + for _, u := range keys { + if wins >= u.n { + _, _ = tx.ExecContext(ctx, `INSERT INTO client_unlocks(client_id,unlock_key,task_id,created_at) VALUES(?,?,?,?) ON CONFLICT(client_id,unlock_key) DO NOTHING`, cid, u.k, taskID, nowMS) + } + } + return nil +} + +func (s *Store) Points(ctx context.Context, taskID string, limit int) ([]Point, error) { + if limit < 1 || limit > 100000 { + limit = 5000 + } + rows, err := s.DB.QueryContext(ctx, `SELECT p.client_id,p.score,p.x,p.y,p.z,p.guess_count,p.last_guess_at,1+(SELECT count(*) FROM task_points p2 WHERE p2.task_id=p.task_id AND p2.score>p.score) rank FROM task_points p WHERE p.task_id=? ORDER BY p.score DESC,p.client_id LIMIT ?`, taskID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]Point, 0) + for rows.Next() { + var p Point + var last sql.NullInt64 + if err := rows.Scan(&p.ClientID, &p.Score, &p.X, &p.Y, &p.Z, &p.GuessCount, &last, &p.Rank); err != nil { + return nil, err + } + if last.Valid { + p.LastGuessAt = fromUnixMS(last.Int64) + } + out = append(out, p) + } + return out, rows.Err() +} + +type Me struct { + ClientID string `json:"client_id"` + Score float64 `json:"score"` + Rank int64 `json:"rank"` + Wins int `json:"wins"` + Unlocks []string `json:"unlocks"` +} + +func (s *Store) Me(ctx context.Context, cid, taskID string) (Me, error) { + m := Me{ClientID: cid, Unlocks: []string{}} + _ = s.DB.QueryRowContext(ctx, `SELECT score,1+(SELECT count(*) FROM task_points p2 WHERE p2.task_id=p.task_id AND p2.score>p.score) FROM task_points p WHERE task_id=? AND client_id=?`, taskID, cid).Scan(&m.Score, &m.Rank) + _ = s.DB.QueryRowContext(ctx, `SELECT count(*) FROM tasks WHERE winner_client_id=? AND status='completed'`, cid).Scan(&m.Wins) + rows, err := s.DB.QueryContext(ctx, `SELECT unlock_key FROM client_unlocks WHERE client_id=? ORDER BY created_at`, cid) + if err == nil { + defer rows.Close() + for rows.Next() { + var u string + _ = rows.Scan(&u) + m.Unlocks = append(m.Unlocks, u) + } + } + return m, nil +} + +type Leader struct { + ClientID string `json:"client_id"` + Wins int `json:"wins"` + BestScore float64 `json:"best_score"` + LiveScore float64 `json:"live_score"` + GuessCount int64 `json:"guess_count"` + Connected bool `json:"connected"` + Unlocks []string `json:"unlocks"` + NFTCount int `json:"nft_count"` + NFTTaskID *string `json:"nft_task_id,omitempty"` + NFTPreviewURI *string `json:"nft_preview_uri,omitempty"` +} + +func (s *Store) Leaderboard(ctx context.Context, limit int) ([]Leader, error) { + return s.leaderboard(ctx, limit, false) +} + +func (s *Store) LiveLeaderboard(ctx context.Context, limit int) ([]Leader, error) { + return s.leaderboard(ctx, limit, true) +} + +func (s *Store) leaderboard(ctx context.Context, limit int, live bool) ([]Leader, error) { + if limit < 1 || limit > 500 { + limit = 100 + } + now := time.Now().UTC().UnixMilli() + order := "wins DESC,best_score DESC,live_score DESC,c.created_at ASC" + if live { + order = "live_score DESC,wins DESC,best_score DESC,c.created_at ASC" + } + q := `SELECT c.id, + (SELECT count(*) FROM tasks t WHERE t.winner_client_id=c.id AND t.status='completed') AS wins, + COALESCE((SELECT max(p.score) FROM task_points p WHERE p.client_id=c.id),0) AS best_score, + COALESCE((SELECT max(p.score) FROM task_points p JOIN tasks t ON t.id=p.task_id WHERE p.client_id=c.id AND t.status='active'),0) AS live_score, + COALESCE((SELECT sum(p.guess_count) FROM task_points p WHERE p.client_id=c.id),0) AS guess_count, + EXISTS(SELECT 1 FROM presence_leases pl WHERE pl.client_id=c.id AND pl.expires_at>?) AS connected, + (SELECT count(*) FROM tasks a WHERE a.winner_client_id=c.id AND a.status='completed' AND a.artifact_status='ready' AND a.artifact_uri IS NOT NULL) AS nft_count, + (SELECT a.id FROM tasks a WHERE a.winner_client_id=c.id AND a.status='completed' AND a.artifact_status='ready' AND a.artifact_uri IS NOT NULL ORDER BY COALESCE(a.completed_at,a.created_at) DESC LIMIT 1) AS nft_task_id + FROM clients c ORDER BY ` + order + ` LIMIT ?` + rows, err := s.DB.QueryContext(ctx, q, now, limit) + if err != nil { + return nil, err + } + out := make([]Leader, 0) + for rows.Next() { + var l Leader + var connected int + var nftTaskID sql.NullString + if err := rows.Scan(&l.ClientID, &l.Wins, &l.BestScore, &l.LiveScore, &l.GuessCount, &connected, &l.NFTCount, &nftTaskID); err != nil { + rows.Close() + return nil, err + } + l.Connected = connected != 0 + l.Unlocks = []string{} + if nftTaskID.Valid { + id := nftTaskID.String + preview := "/api/public/artifacts/" + url.PathEscape(id) + "/preview" + l.NFTTaskID = &id + l.NFTPreviewURI = &preview + } + out = append(out, l) + } + if err := rows.Close(); err != nil { + return nil, err + } + for i := range out { + urows, err := s.DB.QueryContext(ctx, `SELECT unlock_key FROM client_unlocks WHERE client_id=? ORDER BY created_at`, out[i].ClientID) + if err != nil { + return nil, err + } + for urows.Next() { + var u string + if err := urows.Scan(&u); err != nil { + urows.Close() + return nil, err + } + out[i].Unlocks = append(out[i].Unlocks, u) + } + urows.Close() + } + return out, nil +} + +// PublicArtifact is deliberately limited to data that is safe for the public +// leaderboard. The original artifact URI is not exposed here; callers receive +// only the server-generated watermarked preview URL. +type PublicArtifact struct { + TaskID string `json:"task_id"` + WinnerClientID string `json:"winner_client_id"` + RangeBits int `json:"range_bits"` + CompletedAt time.Time `json:"completed_at"` + PreviewURI string `json:"preview_uri"` +} + +func (s *Store) PublicArtifacts(ctx context.Context, limit int, winner string) ([]PublicArtifact, error) { + if limit < 1 || limit > 200 { + limit = 48 + } + winner = strings.TrimSpace(winner) + rows, err := s.DB.QueryContext(ctx, `SELECT id,winner_client_id,range_bits,COALESCE(completed_at,created_at) + FROM tasks + WHERE status='completed' AND artifact_status='ready' AND artifact_uri IS NOT NULL + AND (?='' OR winner_client_id=?) + ORDER BY COALESCE(completed_at,created_at) DESC LIMIT ?`, winner, winner, limit) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]PublicArtifact, 0) + for rows.Next() { + var a PublicArtifact + var completed int64 + if err := rows.Scan(&a.TaskID, &a.WinnerClientID, &a.RangeBits, &completed); err != nil { + return nil, err + } + a.CompletedAt = fromUnixMS(completed) + a.PreviewURI = "/api/public/artifacts/" + url.PathEscape(a.TaskID) + "/preview" + out = append(out, a) + } + return out, rows.Err() +} + +func (s *Store) PublicArtifactSource(ctx context.Context, taskID string) (artifactURI, winner string, ok bool, err error) { + err = s.DB.QueryRowContext(ctx, `SELECT artifact_uri,winner_client_id FROM tasks + WHERE id=? AND status='completed' AND artifact_status='ready' AND artifact_uri IS NOT NULL AND winner_client_id IS NOT NULL`, taskID).Scan(&artifactURI, &winner) + if errors.Is(err, sql.ErrNoRows) { + return "", "", false, nil + } + if err != nil { + return "", "", false, err + } + return artifactURI, winner, true, nil +} + +func (s *Store) TaskArtifactURIs(ctx context.Context, taskID string) (imageURI, manifestURI string, ok bool, err error) { + var image, manifest sql.NullString + err = s.DB.QueryRowContext(ctx, `SELECT artifact_uri,artifact_manifest_uri FROM tasks WHERE id=? AND artifact_status='ready'`, taskID).Scan(&image, &manifest) + if errors.Is(err, sql.ErrNoRows) { + return "", "", false, nil + } + if err != nil { + return "", "", false, err + } + if !image.Valid { + return "", "", false, nil + } + return image.String, manifest.String, true, nil +} + +type AdminTask struct { + ID string `json:"id"` + Status string `json:"status"` + Paused bool `json:"paused"` + RangeBits int `json:"range_bits"` + GuessMinIntervalSec *int `json:"guess_min_interval_sec"` + ClientSubmitIntervalSec *int `json:"client_submit_interval_sec"` + Revision int64 `json:"revision"` + ParentTaskID *string `json:"parent_task_id,omitempty"` + DisplayName string `json:"display_name"` + Description string `json:"description"` + NFTPromptInstructions string `json:"nft_prompt_instructions"` + NFTNegativePrompt string `json:"nft_negative_prompt"` + NFTStyleReference string `json:"nft_style_reference"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at"` + WinnerClientID *string `json:"winner_client_id"` + PointCount int `json:"point_count"` + GuessCount int64 `json:"guess_count"` + ArtifactStatus string `json:"artifact_status"` + ArtifactURI *string `json:"artifact_uri"` + ArtifactManifestURI *string `json:"artifact_manifest_uri"` + ArtifactError *string `json:"artifact_error"` +} + +func (s *Store) AdminTasks(ctx context.Context, status, query string, limit int) ([]AdminTask, error) { + if limit < 1 || limit > 1000 { + limit = 200 + } + rows, err := s.DB.QueryContext(ctx, `SELECT t.id,t.status,t.paused,t.range_bits,t.guess_min_interval_sec,t.client_submit_interval_sec,t.revision,t.parent_task_id,t.display_name,t.description,t.nft_prompt_instructions,t.nft_negative_prompt,t.nft_style_reference,t.created_at,t.completed_at,t.winner_client_id, + (SELECT count(*) FROM task_points p WHERE p.task_id=t.id), + COALESCE((SELECT sum(p.guess_count) FROM task_points p WHERE p.task_id=t.id),0), + t.artifact_status,t.artifact_uri,t.artifact_manifest_uri,t.artifact_error + FROM tasks t + WHERE (?='' OR t.status=?) AND (?='' OR lower(t.id) LIKE '%'||lower(?)||'%' OR lower(COALESCE(t.winner_client_id,'')) LIKE '%'||lower(?)||'%' OR lower(COALESCE(t.display_name,'')) LIKE '%'||lower(?)||'%') + ORDER BY t.created_at DESC LIMIT ?`, status, status, query, query, query, query, limit) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]AdminTask, 0) + for rows.Next() { + var t AdminTask + var paused int + var created int64 + var completed sql.NullInt64 + var winner, artifactURI, manifestURI, artifactErr, parent sql.NullString + var guessMin, clientSubmit sql.NullInt64 + if err := rows.Scan(&t.ID, &t.Status, &paused, &t.RangeBits, &guessMin, &clientSubmit, &t.Revision, &parent, &t.DisplayName, &t.Description, &t.NFTPromptInstructions, &t.NFTNegativePrompt, &t.NFTStyleReference, &created, &completed, &winner, &t.PointCount, &t.GuessCount, &t.ArtifactStatus, &artifactURI, &manifestURI, &artifactErr); err != nil { + return nil, err + } + t.Paused = paused != 0 + if parent.Valid { + v := parent.String + t.ParentTaskID = &v + } + if guessMin.Valid { + v := int(guessMin.Int64) + t.GuessMinIntervalSec = &v + } + if clientSubmit.Valid { + v := int(clientSubmit.Int64) + t.ClientSubmitIntervalSec = &v + } + t.CreatedAt = fromUnixMS(created) + if completed.Valid { + v := fromUnixMS(completed.Int64) + t.CompletedAt = &v + } + if winner.Valid { + v := winner.String + t.WinnerClientID = &v + } + if artifactURI.Valid { + v := artifactURI.String + t.ArtifactURI = &v + } + if manifestURI.Valid { + v := manifestURI.String + t.ArtifactManifestURI = &v + } + if artifactErr.Valid { + v := artifactErr.String + t.ArtifactError = &v + } + out = append(out, t) + } + return out, rows.Err() +} + +func (s *Store) UpdateTaskConfig(ctx context.Context, id, displayName, description, promptInstructions, negativePrompt string) error { + displayName = strings.TrimSpace(displayName) + description = strings.TrimSpace(description) + promptInstructions = strings.TrimSpace(promptInstructions) + negativePrompt = strings.TrimSpace(negativePrompt) + if len(displayName) > 80 { + return fmt.Errorf("display name too long (max 80)") + } + if len(description) > 1200 { + return fmt.Errorf("description too long (max 1200)") + } + if len(promptInstructions) > 8000 { + return fmt.Errorf("NFT prompt instructions too long (max 8000)") + } + if len(negativePrompt) > 4000 { + return fmt.Errorf("NFT negative prompt too long (max 4000)") + } + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET display_name=?,description=?,nft_prompt_instructions=?,nft_negative_prompt=?,revision=revision+1 WHERE id=?`, displayName, description, promptInstructions, negativePrompt, id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("task not found") + } + return nil +} + +func (s *Store) SetTaskStyleReference(ctx context.Context, id, styleReference string) error { + styleReference = strings.TrimSpace(styleReference) + if len(styleReference) > 180 { + return fmt.Errorf("style reference id too long") + } + if styleReference != "" { + // Only opaque basenames created by the admin upload handler are stored. + // Reject path separators so a compromised DB value cannot escape the + // artifact style directory later. + if filepath.Base(styleReference) != styleReference || strings.ContainsAny(styleReference, `/\\`) { + return fmt.Errorf("invalid style reference id") + } + } + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET nft_style_reference=?,revision=revision+1 WHERE id=?`, styleReference, id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("task not found") + } + return nil +} + +func (s *Store) CloseTask(ctx context.Context, id string) error { + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET status='closed',paused=0,completed_at=COALESCE(completed_at,?),revision=revision+1 WHERE id=? AND status='active'`, time.Now().UTC().UnixMilli(), id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("active task not found") + } + return nil +} + +func (s *Store) SetTaskPaused(ctx context.Context, id string, paused bool) error { + v := 0 + if paused { + v = 1 + } + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET paused=?,revision=revision+1 WHERE id=? AND status='active'`, v, id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("active task not found") + } + return nil +} + +func (s *Store) SetTaskIntervals(ctx context.Context, id string, serverSec, clientSec int) error { + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET guess_min_interval_sec=?,client_submit_interval_sec=?,revision=revision+1 WHERE id=? AND status='active'`, serverSec, clientSec, id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("active task not found") + } + return nil +} + +func (s *Store) ClearTaskIntervals(ctx context.Context, id string) error { + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET guess_min_interval_sec=NULL,client_submit_interval_sec=NULL,revision=revision+1 WHERE id=? AND status='active'`, id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("active task not found") + } + return nil +} + +// SetTaskRangeBits changes the live number space. In preserve mode the secret +// stays unchanged; therefore shrinking is only valid while the current secret +// still fits the new range. Existing best scores can be exactly re-expressed +// for the new bit denominator from their stored logarithmic score. Reroll mode +// creates a new secret/seed and resets proximity/sequence while retaining the +// cumulative guess counter. +func (s *Store) SetTaskRangeBits(ctx context.Context, id string, bits int, mode string) error { + if bits < 8 || bits > 128 { + return fmt.Errorf("bits must be 8..128") + } + mode = strings.ToLower(strings.TrimSpace(mode)) + if mode == "" { + mode = "preserve" + } + if mode != "preserve" && mode != "reroll" { + return fmt.Errorf("mode must be preserve or reroll") + } + + var newSecret, newSeed string + var err error + if mode == "reroll" { + newSecret, err = core.RandomDecimal(bits) + if err != nil { + return err + } + newSeed, err = core.RandomSeed() + if err != nil { + return err + } + } + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + var oldBits int + var secret, status string + if err := tx.QueryRowContext(ctx, `SELECT range_bits,secret,status FROM tasks WHERE id=?`, id).Scan(&oldBits, &secret, &status); err != nil { + return err + } + if status != "active" { + return fmt.Errorf("active task not found") + } + if mode == "preserve" { + sv, ok := new(big.Int).SetString(secret, 10) + if !ok { + return fmt.Errorf("invalid stored secret") + } + max := new(big.Int).Lsh(big.NewInt(1), uint(bits)) + if sv.Cmp(max) >= 0 { + return fmt.Errorf("current secret does not fit %d bits; use reroll mode", bits) + } + type pointScore struct { + cid string + score float64 + } + rows, err := tx.QueryContext(ctx, `SELECT client_id,score FROM task_points WHERE task_id=?`, id) + if err != nil { + return err + } + var points []pointScore + for rows.Next() { + var p pointScore + if err := rows.Scan(&p.cid, &p.score); err != nil { + rows.Close() + return err + } + points = append(points, p) + } + if err := rows.Close(); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `UPDATE tasks SET range_bits=?,revision=revision+1 WHERE id=?`, bits, id); err != nil { + return err + } + for _, p := range points { + missLog := float64(oldBits) * (1 - p.score/100) + newScore := 100 * (1 - missLog/float64(bits)) + newScore = math.Max(0, math.Min(100, newScore)) + x, y, z := core.Position(p.cid, newScore) + if _, err := tx.ExecContext(ctx, `UPDATE task_points SET score=?,x=?,y=?,z=? WHERE task_id=? AND client_id=?`, newScore, x, y, z, id, p.cid); err != nil { + return err + } + } + } else { + if _, err := tx.ExecContext(ctx, `UPDATE tasks SET range_bits=?,secret=?,public_seed=?,revision=revision+1 WHERE id=?`, bits, newSecret, newSeed, id); err != nil { + return err + } + rows, err := tx.QueryContext(ctx, `SELECT client_id FROM task_points WHERE task_id=?`, id) + if err != nil { + return err + } + var cids []string + for rows.Next() { + var cid string + if err := rows.Scan(&cid); err != nil { + rows.Close() + return err + } + cids = append(cids, cid) + } + if err := rows.Close(); err != nil { + return err + } + for _, cid := range cids { + x, y, z := core.Position(cid, 0) + if _, err := tx.ExecContext(ctx, `UPDATE task_points SET score=0,x=?,y=?,z=?,next_seq=0,last_guess_at=NULL WHERE task_id=? AND client_id=?`, x, y, z, id, cid); err != nil { + return err + } + } + } + return tx.Commit() +} + +func (s *Store) RerollTask(ctx context.Context, id string) error { + var bits int + if err := s.DB.QueryRowContext(ctx, `SELECT range_bits FROM tasks WHERE id=?`, id).Scan(&bits); err != nil { + return err + } + return s.SetTaskRangeBits(ctx, id, bits, "reroll") +} + +func (s *Store) QueueArtifact(ctx context.Context, id string) error { + res, err := s.DB.ExecContext(ctx, `UPDATE tasks SET artifact_status='pending',artifact_uri=NULL,artifact_manifest_uri=NULL,artifact_error=NULL WHERE id=? AND status='completed'`, id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("completed task not found") + } + return nil +} + +type TaskAction struct { + ID string `json:"id"` + TaskID string `json:"task_id"` + ActionType string `json:"action_type"` + Payload json.RawMessage `json:"payload"` + ExecuteAt time.Time `json:"execute_at"` + Status string `json:"status"` + CreatedAt time.Time `json:"created_at"` + ExecutedAt *time.Time `json:"executed_at"` + Error *string `json:"error"` +} + +func (s *Store) ScheduleTaskAction(ctx context.Context, taskID, actionType string, payload any, executeAt time.Time) (TaskAction, error) { + if executeAt.IsZero() { + executeAt = time.Now().UTC() + } + b, err := json.Marshal(payload) + if err != nil { + return TaskAction{}, err + } + a := TaskAction{ID: NewID("act_"), TaskID: taskID, ActionType: actionType, Payload: b, ExecuteAt: executeAt.UTC(), Status: "pending", CreatedAt: time.Now().UTC()} + _, err = s.DB.ExecContext(ctx, `INSERT INTO task_actions(id,task_id,action_type,payload_json,execute_at,status,created_at) VALUES(?,?,?,?,?,'pending',?)`, a.ID, a.TaskID, a.ActionType, string(b), a.ExecuteAt.UnixMilli(), a.CreatedAt.UnixMilli()) + return a, err +} + +func scanAction(scanner interface{ Scan(...any) error }) (TaskAction, error) { + var a TaskAction + var raw string + var executeAt, createdAt int64 + var executed sql.NullInt64 + var errText sql.NullString + if err := scanner.Scan(&a.ID, &a.TaskID, &a.ActionType, &raw, &executeAt, &a.Status, &createdAt, &executed, &errText); err != nil { + return TaskAction{}, err + } + a.Payload = json.RawMessage(raw) + a.ExecuteAt = fromUnixMS(executeAt) + a.CreatedAt = fromUnixMS(createdAt) + if executed.Valid { + v := fromUnixMS(executed.Int64) + a.ExecutedAt = &v + } + if errText.Valid { + v := errText.String + a.Error = &v + } + return a, nil +} + +func (s *Store) TaskActions(ctx context.Context, taskID string, limit int) ([]TaskAction, error) { + if limit < 1 || limit > 500 { + limit = 100 + } + rows, err := s.DB.QueryContext(ctx, `SELECT id,task_id,action_type,payload_json,execute_at,status,created_at,executed_at,error FROM task_actions WHERE task_id=? ORDER BY created_at DESC LIMIT ?`, taskID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]TaskAction, 0) + for rows.Next() { + a, err := scanAction(rows) + if err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func (s *Store) DueTaskActions(ctx context.Context, limit int) ([]TaskAction, error) { + if limit < 1 || limit > 100 { + limit = 20 + } + rows, err := s.DB.QueryContext(ctx, `SELECT id,task_id,action_type,payload_json,execute_at,status,created_at,executed_at,error FROM task_actions WHERE status='pending' AND execute_at<=? ORDER BY execute_at,id LIMIT ?`, time.Now().UTC().UnixMilli(), limit) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]TaskAction, 0) + for rows.Next() { + a, err := scanAction(rows) + if err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func (s *Store) StartTaskAction(ctx context.Context, id string) bool { + res, err := s.DB.ExecContext(ctx, `UPDATE task_actions SET status='running',error=NULL WHERE id=? AND status='pending'`, id) + if err != nil { + return false + } + n, _ := res.RowsAffected() + return n == 1 +} + +func (s *Store) FinishTaskAction(ctx context.Context, id string, runErr error) { + now := time.Now().UTC().UnixMilli() + if runErr == nil { + _, _ = s.DB.ExecContext(ctx, `UPDATE task_actions SET status='done',executed_at=?,error=NULL WHERE id=?`, now, id) + return + } + _, _ = s.DB.ExecContext(ctx, `UPDATE task_actions SET status='error',executed_at=?,error=? WHERE id=?`, now, runErr.Error(), id) +} + +func (s *Store) CancelTaskAction(ctx context.Context, id string) error { + res, err := s.DB.ExecContext(ctx, `UPDATE task_actions SET status='cancelled',executed_at=? WHERE id=? AND status='pending'`, time.Now().UTC().UnixMilli(), id) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return fmt.Errorf("pending action not found") + } + return nil +} + +// ProfileCleanupCandidate is an old client identity that has never won a task. +// Runtime connection state is deliberately checked by the server because the +// hot-path presence registry is process-local, not stored in SQLite. +type ProfileCleanupCandidate struct { + ClientID string `json:"client_id"` + LastSeen int64 `json:"last_seen"` +} + +// InactiveNonWinnerClients returns identities whose persisted last activity is +// older than cutoffMS and that have never been recorded as a task winner. +func (s *Store) InactiveNonWinnerClients(ctx context.Context, cutoffMS int64) ([]ProfileCleanupCandidate, error) { + rows, err := s.DB.QueryContext(ctx, `SELECT c.id,c.last_seen + FROM clients c + WHERE c.last_seen < ? + AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.winner_client_id=c.id) + ORDER BY c.last_seen ASC`, cutoffMS) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]ProfileCleanupCandidate, 0) + for rows.Next() { + var c ProfileCleanupCandidate + if err := rows.Scan(&c.ClientID, &c.LastSeen); err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +// OldWinnerCount reports old identities protected from cleanup because they +// have won at least one task. Winners are never deleted by profile cleanup. +func (s *Store) OldWinnerCount(ctx context.Context, cutoffMS int64) (int64, error) { + var n int64 + err := s.DB.QueryRowContext(ctx, `SELECT count(*) FROM clients c + WHERE c.last_seen < ? + AND EXISTS (SELECT 1 FROM tasks t WHERE t.winner_client_id=c.id)`, cutoffMS).Scan(&n) + return n, err +} + +// DeleteInactiveNonWinnerClients deletes only the explicitly supplied client +// IDs and re-checks both age and winner protection inside the transaction. The +// re-check makes a concurrent login safe because login/WS connect refreshes +// clients.last_seen before this DELETE can match it. +func (s *Store) DeleteInactiveNonWinnerClients(ctx context.Context, cutoffMS int64, ids []string) ([]string, error) { + if len(ids) == 0 { + return []string{}, nil + } + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + deleted := make([]string, 0, len(ids)) + for _, id := range ids { + id = strings.TrimSpace(id) + if id == "" { + continue + } + // presence_leases is legacy/ephemeral and has no FK; remove any stale row. + _, _ = tx.ExecContext(ctx, `DELETE FROM presence_leases WHERE client_id=? AND expires_at<=?`, id, time.Now().UTC().UnixMilli()) + res, err := tx.ExecContext(ctx, `DELETE FROM clients + WHERE id=? AND last_seen < ? + AND NOT EXISTS (SELECT 1 FROM tasks t WHERE t.winner_client_id=clients.id)`, id, cutoffMS) + if err != nil { + return nil, err + } + n, err := res.RowsAffected() + if err != nil { + return nil, err + } + if n == 1 { + deleted = append(deleted, id) + } + } + if err := tx.Commit(); err != nil { + return nil, err + } + return deleted, nil +} + +// TouchClient marks authenticated client activity without modifying identity +// material. It is used at WebSocket connect/disconnect so cleanup reflects +// actual recent use rather than only the last login time. +func (s *Store) TouchClient(ctx context.Context, id string) { + _, _ = s.DB.ExecContext(ctx, `UPDATE clients SET last_seen=? WHERE id=?`, time.Now().UTC().UnixMilli(), id) +} + +// Presence lease helpers replace Redis SET NX + TTL. +func (s *Store) AcquirePresence(ctx context.Context, cid, sessionID string, ttl time.Duration) error { + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + now := time.Now().UTC().UnixMilli() + var current string + var expires int64 + err = tx.QueryRowContext(ctx, `SELECT session_id,expires_at FROM presence_leases WHERE client_id=?`, cid).Scan(¤t, &expires) + if err == nil && expires > now { + return ErrPresenceBusy + } + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO presence_leases(client_id,session_id,expires_at) VALUES(?,?,?) + ON CONFLICT(client_id) DO UPDATE SET session_id=excluded.session_id,expires_at=excluded.expires_at`, cid, sessionID, now+ttl.Milliseconds()) + if err != nil { + return err + } + return tx.Commit() +} + +func (s *Store) RefreshPresence(ctx context.Context, cid, sessionID string, ttl time.Duration) bool { + now := time.Now().UTC().UnixMilli() + res, err := s.DB.ExecContext(ctx, `UPDATE presence_leases SET expires_at=? WHERE client_id=? AND session_id=? AND expires_at>?`, now+ttl.Milliseconds(), cid, sessionID, now) + if err != nil { + return false + } + n, _ := res.RowsAffected() + return n == 1 +} + +func (s *Store) ReleasePresence(ctx context.Context, cid, sessionID string) { + _, _ = s.DB.ExecContext(ctx, `DELETE FROM presence_leases WHERE client_id=? AND session_id=?`, cid, sessionID) +} + +func (s *Store) HasPresence(ctx context.Context, cid, sessionID string) bool { + var n int + now := time.Now().UTC().UnixMilli() + _ = s.DB.QueryRowContext(ctx, `SELECT count(*) FROM presence_leases WHERE client_id=? AND session_id=? AND expires_at>?`, cid, sessionID, now).Scan(&n) + return n == 1 +} + +func (s *Store) ConnectedCount(ctx context.Context) int64 { + var n int64 + now := time.Now().UTC().UnixMilli() + _ = s.DB.QueryRowContext(ctx, `SELECT count(*) FROM presence_leases WHERE expires_at>?`, now).Scan(&n) + return n +} + +func (s *Store) CleanupEphemeral(ctx context.Context) { + now := time.Now().UTC().UnixMilli() + _, _ = s.DB.ExecContext(ctx, `DELETE FROM presence_leases WHERE expires_at<=?`, now) +} + +func IsNoRows(err error) bool { return errors.Is(err, sql.ErrNoRows) } + +// GuessStateSnapshot is the persisted checkpoint used to seed the in-memory +// hot-path state. Normal losing guesses are intentionally not persisted. +type GuessStateSnapshot struct { + NextSeq int64 + LastGuess time.Time + BestScore float64 + GuessCount int64 +} + +func (s *Store) LoadGuessState(ctx context.Context, taskID, cid string) (GuessStateSnapshot, error) { + var out GuessStateSnapshot + var last sql.NullInt64 + err := s.DB.QueryRowContext(ctx, `SELECT next_seq,last_guess_at,score,guess_count FROM task_points WHERE task_id=? AND client_id=?`, taskID, cid). + Scan(&out.NextSeq, &last, &out.BestScore, &out.GuessCount) + if errors.Is(err, sql.ErrNoRows) { + return out, nil + } + if err != nil { + return out, err + } + if last.Valid { + out.LastGuess = fromUnixMS(last.Int64) + } + return out, nil +} + +// PersistImprovement checkpoints only meaningful state changes. The absolute +// sequence/count values include all losing guesses that happened in memory +// since the previous checkpoint, so a restart resumes from the latest durable +// improvement rather than writing every false guess. +func (s *Store) PersistImprovement(ctx context.Context, t SecretTask, cid string, nextSeq, guessCount int64, lastGuess time.Time, score float64, guess, sig string, correct bool) (Point, error) { + tx, err := s.DB.BeginTx(ctx, nil) + if err != nil { + return Point{}, err + } + defer tx.Rollback() + var status string + var paused int + if err := tx.QueryRowContext(ctx, `SELECT status,paused FROM tasks WHERE id=?`, t.ID).Scan(&status, &paused); err != nil { + return Point{}, err + } + if status != "active" { + return Point{}, ErrTaskCompleted + } + if paused != 0 { + return Point{}, ErrTaskPaused + } + x, y, z := core.Position(cid, score) + lastMS := lastGuess.UTC().UnixMilli() + _, err = tx.ExecContext(ctx, `INSERT INTO task_points(task_id,client_id,score,x,y,z,guess_count,next_seq,last_guess_at) + VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(task_id,client_id) DO UPDATE SET + score=excluded.score,x=excluded.x,y=excluded.y,z=excluded.z,guess_count=excluded.guess_count,next_seq=excluded.next_seq,last_guess_at=excluded.last_guess_at`, + t.ID, cid, score, x, y, z, guessCount, nextSeq, lastMS) + if err != nil { + return Point{}, err + } + if correct { + res, err := tx.ExecContext(ctx, `UPDATE tasks SET status='completed',completed_at=?,winner_client_id=?,winner_signature=?,winning_guess=?,artifact_status='pending',revision=revision+1 WHERE id=? AND status='active'`, lastMS, cid, sig, guess, t.ID) + if err != nil { + return Point{}, err + } + n, _ := res.RowsAffected() + if n != 1 { + return Point{}, ErrTaskCompleted + } + if err := s.unlocksTx(ctx, tx, cid, t.ID, lastMS); err != nil { + return Point{}, err + } + } + var p Point + p.ClientID = cid + p.Score = score + p.X = x + p.Y = y + p.Z = z + p.GuessCount = guessCount + p.LastGuessAt = lastGuess + if err := tx.QueryRowContext(ctx, `SELECT 1+count(*) FROM task_points WHERE task_id=? AND score>?`, t.ID, score).Scan(&p.Rank); err != nil { + return Point{}, err + } + if err := tx.Commit(); err != nil { + return Point{}, err + } + return p, nil +} + +// PointsForClient bounds map snapshots server-side while guaranteeing that the +// requesting client is present even when it is not in the global top-N. +func (s *Store) PointsForClient(ctx context.Context, taskID, cid string, limit int) ([]Point, error) { + if limit < 10 { + limit = 10 + } + if limit > 10000 { + limit = 10000 + } + ps, err := s.Points(ctx, taskID, limit) + if err != nil { + return nil, err + } + for _, p := range ps { + if p.ClientID == cid { + return ps, nil + } + } + var own Point + var last sql.NullInt64 + err = s.DB.QueryRowContext(ctx, `SELECT p.client_id,p.score,p.x,p.y,p.z,p.guess_count,p.last_guess_at,1+(SELECT count(*) FROM task_points p2 WHERE p2.task_id=p.task_id AND p2.score>p.score) FROM task_points p WHERE p.task_id=? AND p.client_id=?`, taskID, cid). + Scan(&own.ClientID, &own.Score, &own.X, &own.Y, &own.Z, &own.GuessCount, &last, &own.Rank) + if errors.Is(err, sql.ErrNoRows) { + return ps, nil + } + if err != nil { + return nil, err + } + if last.Valid { + own.LastGuessAt = fromUnixMS(last.Int64) + } + if len(ps) >= limit { + ps = ps[:limit-1] + } + return append(ps, own), nil +} diff --git a/internal/data/successor_test.go b/internal/data/successor_test.go new file mode 100644 index 0000000..9638e92 --- /dev/null +++ b/internal/data/successor_test.go @@ -0,0 +1,68 @@ +package data + +import ( + "context" + "testing" + + "neuralhunt/internal/auth" +) + +func TestSuccessorInheritsTaskConfigAndSelection(t *testing.T) { + ctx := context.Background() + db, err := OpenSQLite(ctx, t.TempDir()+"/test.db") + if err != nil { + t.Fatal(err) + } + defer db.Close() + s := New(db) + if err := s.EnsureActiveTasks(ctx, 1, 40); err != nil { + t.Fatal(err) + } + ts, err := s.ActiveTasks(ctx) + if err != nil || len(ts) != 1 { + t.Fatalf("active tasks: len=%d err=%v", len(ts), err) + } + pred := ts[0] + if err := s.UpdateTaskConfig(ctx, pred.ID, "Prime Field", "Beschreibung", "winner in crystalline neural lattice", "no text"); err != nil { + t.Fatal(err) + } + if err := s.SetTaskStyleReference(ctx, pred.ID, "abc123.jpg"); err != nil { + t.Fatal(err) + } + if err := s.SetTaskIntervals(ctx, pred.ID, 13, 14); err != nil { + t.Fatal(err) + } + jwk := auth.PublicJWK{Kty: "EC", Crv: "P-256", X: "AQ", Y: "Ag"} + // UpsertClient stores the JWK without validating its curve; crypto validation + // belongs to auth handlers and is irrelevant to this persistence test. + if err := s.UpsertClient(ctx, "client_test", jwk); err != nil { + t.Fatal(err) + } + if err := s.SetClientTaskSelection(ctx, "client_test", pred.ID); err != nil { + t.Fatal(err) + } + if err := s.CloseTask(ctx, pred.ID); err != nil { + t.Fatal(err) + } + next, err := s.EnsureSuccessorTask(ctx, pred.ID, 32) + if err != nil { + t.Fatal(err) + } + if next.ParentTaskID == nil || *next.ParentTaskID != pred.ID { + t.Fatalf("parent mismatch: %#v", next.ParentTaskID) + } + if next.RangeBits != 40 || next.DisplayName != "Prime Field" || next.Description != "Beschreibung" || next.NFTPromptInstructions != "winner in crystalline neural lattice" || next.NFTNegativePrompt != "no text" || next.NFTStyleReference != "abc123.jpg" { + t.Fatalf("successor did not inherit config: %+v", next) + } + if next.GuessMinIntervalSec == nil || *next.GuessMinIntervalSec != 13 || next.ClientSubmitIntervalSec == nil || *next.ClientSubmitIntervalSec != 14 { + t.Fatalf("successor did not inherit intervals: %+v", next) + } + selected, err := s.SelectedTaskID(ctx, "client_test") + if err != nil || selected != next.ID { + t.Fatalf("selection was not migrated: selected=%q next=%q err=%v", selected, next.ID, err) + } + again, err := s.EnsureSuccessorTask(ctx, pred.ID, 32) + if err != nil || again.ID != next.ID { + t.Fatalf("successor must be idempotent: first=%q second=%q err=%v", next.ID, again.ID, err) + } +}