package sqlstore import ( "context" "database/sql" "encoding/json" "errors" "fmt" "path/filepath" "strings" "sync" "time" "gitea-codex-bot/internal/config" "gitea-codex-bot/internal/domain" "gitea-codex-bot/internal/store" _ "github.com/go-sql-driver/mysql" _ "modernc.org/sqlite" ) const defaultMaxRetries = 2 var ErrStaleRun = errors.New("review run is no longer current") type Store struct { db *sql.DB dialect string enqueueMu sync.Mutex } func Open(settings config.Settings) (*Store, error) { dsn := settings.DatabaseURL dialect := "mysql" if dsn == "" { dsn = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true&charset=utf8mb4", settings.DBUser, settings.DBPassword, settings.DBHost, settings.DBPort, settings.DBName) } if strings.HasPrefix(dsn, "mysql://") { dsn = strings.TrimPrefix(dsn, "mysql://") } if strings.HasPrefix(dsn, "sqlite://") { dialect = "sqlite" dsn = strings.TrimPrefix(dsn, "sqlite://") } else if strings.HasPrefix(dsn, "sqlite:") { dialect = "sqlite" dsn = strings.TrimPrefix(dsn, "sqlite:") } else if strings.HasPrefix(dsn, "file:") || dsn == ":memory:" || filepath.Ext(dsn) == ".db" { dialect = "sqlite" } if dialect == "mysql" && !strings.Contains(dsn, "parseTime=") { if strings.Contains(dsn, "?") { dsn += "&parseTime=true" } else { dsn += "?parseTime=true" } } db, err := sql.Open(map[string]string{"sqlite": "sqlite", "mysql": "mysql"}[dialect], dsn) if err != nil { return nil, err } if dialect == "sqlite" { // SQLite has one writer and in-memory databases are connection-local. db.SetMaxOpenConns(1) db.SetMaxIdleConns(1) } else { db.SetMaxOpenConns(8) } if err := db.Ping(); err != nil { _ = db.Close() return nil, err } return &Store{db: db, dialect: dialect}, nil } func (s *Store) Close() error { return s.db.Close() } func (s *Store) Migrate(ctx context.Context) error { stmts := s.schema() for _, stmt := range stmts { if _, err := s.db.ExecContext(ctx, stmt); err != nil { return fmt.Errorf("migration: %w", err) } } if _, err := s.db.ExecContext(ctx, s.alterAddTriggerCommentBody()); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate") && !strings.Contains(strings.ToLower(err.Error()), "exists") { return fmt.Errorf("migration trigger_comment_body: %w", err) } return nil } func (s *Store) schema() []string { if s.dialect == "sqlite" { return []string{ `CREATE TABLE IF NOT EXISTS webhook_events (id INTEGER PRIMARY KEY AUTOINCREMENT, delivery_id TEXT NULL UNIQUE, event_name TEXT NOT NULL, repo TEXT NOT NULL, comment_id INTEGER NULL, payload_sha256 TEXT NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(repo, comment_id))`, `CREATE TABLE IF NOT EXISTS review_jobs (id INTEGER PRIMARY KEY AUTOINCREMENT, repo TEXT NOT NULL, pr_number INTEGER NOT NULL, head_sha TEXT NOT NULL, trigger_comment_id INTEGER NOT NULL, command TEXT NOT NULL, command_args TEXT NULL, requested_by TEXT NOT NULL, status TEXT NOT NULL, last_error TEXT NULL, result_json TEXT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, started_at TIMESTAMP NULL, finished_at TIMESTAMP NULL, UNIQUE(repo, trigger_comment_id))`, `CREATE INDEX IF NOT EXISTS ix_review_jobs_lookup ON review_jobs(repo, pr_number, head_sha, status, created_at)`, `CREATE TABLE IF NOT EXISTS review_runs (id INTEGER PRIMARY KEY AUTOINCREMENT, job_id INTEGER NOT NULL, status TEXT NOT NULL, runner_container_id TEXT NULL, result_json TEXT NULL, error_message TEXT NULL, started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, finished_at TIMESTAMP NULL, FOREIGN KEY(job_id) REFERENCES review_jobs(id) ON DELETE CASCADE)`, `CREATE INDEX IF NOT EXISTS ix_review_runs_job_status ON review_runs(job_id, status)`, `CREATE TABLE IF NOT EXISTS bot_comments (id INTEGER PRIMARY KEY AUTOINCREMENT, repo TEXT NOT NULL, pr_number INTEGER NOT NULL, head_sha TEXT NOT NULL, gitea_comment_id INTEGER NOT NULL, marker TEXT NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(repo, pr_number, marker))`, `CREATE INDEX IF NOT EXISTS ix_bot_comments_repo_pr ON bot_comments(repo, pr_number)`, } } return []string{ `CREATE TABLE IF NOT EXISTS webhook_events (id BIGINT AUTO_INCREMENT PRIMARY KEY, delivery_id VARCHAR(255) NULL UNIQUE, event_name VARCHAR(128) NOT NULL, repo VARCHAR(255) NOT NULL, comment_id BIGINT NULL, payload_sha256 VARCHAR(64) NOT NULL, created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), UNIQUE KEY uq_webhook_events_repo_comment (repo, comment_id)) ENGINE=InnoDB`, `CREATE TABLE IF NOT EXISTS review_jobs (id BIGINT AUTO_INCREMENT PRIMARY KEY, repo VARCHAR(255) NOT NULL, pr_number INT NOT NULL, head_sha VARCHAR(64) NOT NULL, trigger_comment_id BIGINT NOT NULL, command VARCHAR(64) NOT NULL, command_args TEXT NULL, requested_by VARCHAR(255) NOT NULL, status VARCHAR(32) NOT NULL, last_error TEXT NULL, result_json JSON NULL, created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), started_at DATETIME(6) NULL, finished_at DATETIME(6) NULL, UNIQUE KEY uq_review_jobs_repo_trigger_comment (repo, trigger_comment_id), KEY ix_review_jobs_lookup (repo, pr_number, head_sha, status, created_at)) ENGINE=InnoDB`, `CREATE TABLE IF NOT EXISTS review_runs (id BIGINT AUTO_INCREMENT PRIMARY KEY, job_id BIGINT NOT NULL, status VARCHAR(32) NOT NULL, runner_container_id VARCHAR(128) NULL, result_json JSON NULL, error_message TEXT NULL, started_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), finished_at DATETIME(6) NULL, CONSTRAINT fk_review_runs_job FOREIGN KEY(job_id) REFERENCES review_jobs(id) ON DELETE CASCADE, KEY ix_review_runs_job_status (job_id, status)) ENGINE=InnoDB`, `CREATE TABLE IF NOT EXISTS bot_comments (id BIGINT AUTO_INCREMENT PRIMARY KEY, repo VARCHAR(255) NOT NULL, pr_number INT NOT NULL, head_sha VARCHAR(64) NOT NULL, gitea_comment_id BIGINT NOT NULL, marker VARCHAR(255) NOT NULL, created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), UNIQUE KEY uq_bot_comments_marker (repo, pr_number, marker), KEY ix_bot_comments_repo_pr (repo, pr_number)) ENGINE=InnoDB`, } } func (s *Store) alterAddTriggerCommentBody() string { if s.dialect == "sqlite" { return `ALTER TABLE review_jobs ADD COLUMN trigger_comment_body TEXT` } return `ALTER TABLE review_jobs ADD COLUMN trigger_comment_body TEXT NULL` } func (s *Store) InsertWebhookEvent(ctx context.Context, event domain.WebhookEvent) (bool, error) { _, err := s.db.ExecContext(ctx, `INSERT INTO webhook_events(delivery_id,event_name,repo,comment_id,payload_sha256) VALUES(?,?,?,?,?)`, nullable(event.DeliveryID), event.EventName, event.Repo, event.CommentID, event.PayloadSHA256) if err != nil { if isConstraint(err) { return false, nil } return false, err } return true, nil } func (s *Store) CooldownRemaining(ctx context.Context, repo string, pr int, duration time.Duration) (int, error) { cutoff := time.Now().UTC().Add(-duration) var created time.Time err := s.db.QueryRowContext(ctx, `SELECT created_at FROM review_jobs WHERE repo=? AND pr_number=? AND created_at>=? ORDER BY created_at DESC LIMIT 1`, repo, pr, cutoff).Scan(&created) if errors.Is(err, sql.ErrNoRows) { return 0, nil } if err != nil { return 0, err } remaining := int((duration - time.Since(created)).Seconds()) if remaining < 0 { return 0, nil } return remaining, nil } func (s *Store) EnqueueJob(ctx context.Context, event domain.WebhookEvent, command domain.ParsedCommand) (domain.Job, error) { args, _ := json.Marshal(command.Arguments) result, err := s.db.ExecContext(ctx, `INSERT INTO review_jobs(repo,pr_number,head_sha,trigger_comment_id,command,command_args,trigger_comment_body,requested_by,status) VALUES(?,?,?,?,?,?,?,?,?)`, event.Repo, event.PRNumber, event.HeadSHA, event.CommentID, command.Name, string(args), event.CommentBody, event.Sender, domain.JobQueued) if err != nil { return domain.Job{}, err } id, err := result.LastInsertId() if err != nil { return domain.Job{}, err } return s.getJob(ctx, id) } func (s *Store) EnqueueAcceptedJob(ctx context.Context, event domain.WebhookEvent, command domain.ParsedCommand, cooldown time.Duration) (domain.Job, bool, int, error) { s.enqueueMu.Lock() defer s.enqueueMu.Unlock() tx, err := s.db.BeginTx(ctx, nil) if err != nil { return domain.Job{}, false, 0, err } defer tx.Rollback() _, err = tx.ExecContext(ctx, `INSERT INTO webhook_events(delivery_id,event_name,repo,comment_id,payload_sha256) VALUES(?,?,?,?,?)`, nullable(event.DeliveryID), event.EventName, event.Repo, event.CommentID, event.PayloadSHA256) if err != nil { if isConstraint(err) { return domain.Job{}, false, 0, nil } return domain.Job{}, false, 0, err } if command.Name == "review" && cooldown > 0 { cutoff := time.Now().UTC().Add(-cooldown) query := `SELECT created_at FROM review_jobs WHERE repo=? AND pr_number=? AND created_at>=? ORDER BY created_at DESC LIMIT 1` if s.dialect == "mysql" { query += ` FOR UPDATE` } var created time.Time if err := tx.QueryRowContext(ctx, query, event.Repo, event.PRNumber, cutoff).Scan(&created); err == nil { remaining := int((cooldown - time.Since(created)).Seconds()) if remaining < 0 { remaining = 0 } if err := tx.Commit(); err != nil { return domain.Job{}, false, 0, err } return domain.Job{}, true, remaining, nil } else if !errors.Is(err, sql.ErrNoRows) { return domain.Job{}, false, 0, err } } args, _ := json.Marshal(command.Arguments) result, err := tx.ExecContext(ctx, `INSERT INTO review_jobs(repo,pr_number,head_sha,trigger_comment_id,command,command_args,trigger_comment_body,requested_by,status) VALUES(?,?,?,?,?,?,?,?,?)`, event.Repo, event.PRNumber, event.HeadSHA, event.CommentID, command.Name, string(args), event.CommentBody, event.Sender, domain.JobQueued) if err != nil { return domain.Job{}, false, 0, err } id, err := result.LastInsertId() if err != nil { return domain.Job{}, false, 0, err } if err := tx.Commit(); err != nil { return domain.Job{}, false, 0, err } job, err := s.getJob(ctx, id) return job, true, 0, err } func (s *Store) ClaimNextJob(ctx context.Context, now time.Time, lease time.Duration, maxRetries int) (*domain.Job, *domain.ReviewRun, error) { if err := s.recoverStale(ctx, now, lease, maxRetries); err != nil { return nil, nil, err } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return nil, nil, err } defer tx.Rollback() query := `SELECT id FROM review_jobs WHERE status=? ORDER BY created_at ASC,id ASC LIMIT 1` if s.dialect == "mysql" { query += ` FOR UPDATE SKIP LOCKED` } var id int64 if err := tx.QueryRowContext(ctx, query, domain.JobQueued).Scan(&id); err != nil { if errors.Is(err, sql.ErrNoRows) { return nil, nil, nil } return nil, nil, err } started := now.UTC() if _, err := tx.ExecContext(ctx, `UPDATE review_jobs SET status=?, started_at=?, finished_at=NULL, updated_at=? WHERE id=?`, domain.JobRunning, started, started, id); err != nil { return nil, nil, err } runRes, err := tx.ExecContext(ctx, `INSERT INTO review_runs(job_id,status,started_at) VALUES(?,?,?)`, id, domain.RunRunning, started) if err != nil { return nil, nil, err } runID, err := runRes.LastInsertId() if err != nil { return nil, nil, err } if err := tx.Commit(); err != nil { return nil, nil, err } job, err := s.getJob(ctx, id) if err != nil { return nil, nil, err } return &job, &domain.ReviewRun{ID: runID, JobID: id, Status: domain.RunRunning, StartedAt: started}, nil } func (s *Store) recoverStale(ctx context.Context, now time.Time, lease time.Duration, maxRetries int) error { rows, err := s.db.QueryContext(ctx, `SELECT id,started_at FROM review_jobs WHERE status=? AND started_at IS NOT NULL AND started_at<=?`, domain.JobRunning, now.Add(-lease)) if err != nil { return err } defer rows.Close() for rows.Next() { var id int64 var started time.Time if err := rows.Scan(&id, &started); err != nil { return err } var attempts int if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM review_runs WHERE job_id=?`, id).Scan(&attempts); err != nil { return err } message := fmt.Sprintf("Job lease timed out after %ds on attempt %d. Recovered by queue watchdog.", int(lease.Seconds()), attempts) _, _ = s.db.ExecContext(ctx, `UPDATE review_runs SET status=?,finished_at=?,error_message=? WHERE id=(SELECT id FROM (SELECT id FROM review_runs WHERE job_id=? ORDER BY id DESC LIMIT 1) AS latest) AND status=?`, domain.RunFailed, now, message, id, domain.RunRunning) if attempts-1 < maxRetries { _, err = s.db.ExecContext(ctx, `UPDATE review_jobs SET status=?,started_at=NULL,finished_at=NULL,last_error=?,updated_at=? WHERE id=?`, domain.JobQueued, message, now, id) } else { _, err = s.db.ExecContext(ctx, `UPDATE review_jobs SET status=?,finished_at=?,last_error=?,updated_at=? WHERE id=?`, domain.JobFailed, now, message, now, id) } if err != nil { return err } } return rows.Err() } func (s *Store) FinishJob(ctx context.Context, jobID, runID int64, success, skipped bool, result *domain.ReviewResult, runErr error) error { now := time.Now().UTC() status, runStatus := domain.JobFailed, domain.RunFailed if skipped { status, runStatus = domain.JobSkipped, domain.RunSkipped } else if success { status, runStatus = domain.JobSucceeded, domain.RunSucceeded } else { var attempts int if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM review_runs WHERE job_id=?`, jobID).Scan(&attempts); err != nil { return err } if attempts-1 < defaultMaxRetries { status = domain.JobQueued } } var resultJSON []byte if result != nil { resultJSON, _ = json.Marshal(result) } errText := "" if runErr != nil { errText = runErr.Error() } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() runUpdate, err := tx.ExecContext(ctx, `UPDATE review_runs SET status=?,finished_at=?,error_message=?,result_json=? WHERE id=? AND status=?`, runStatus, now, nullable(errText), nullableBytes(resultJSON), runID, domain.RunRunning) if err != nil { return err } if affected, affectedErr := runUpdate.RowsAffected(); affectedErr != nil || affected == 0 { return ErrStaleRun } var jobUpdate sql.Result if status == domain.JobQueued { jobUpdate, err = tx.ExecContext(ctx, `UPDATE review_jobs SET status=?,started_at=NULL,finished_at=NULL,last_error=?,result_json=?,updated_at=? WHERE id=? AND status=?`, status, nullable(errText), nullableBytes(resultJSON), now, jobID, domain.JobRunning) } else { jobUpdate, err = tx.ExecContext(ctx, `UPDATE review_jobs SET status=?,finished_at=?,last_error=?,result_json=?,updated_at=? WHERE id=? AND status=?`, status, now, nullable(errText), nullableBytes(resultJSON), now, jobID, domain.JobRunning) } if err != nil { return err } if affected, affectedErr := jobUpdate.RowsAffected(); affectedErr != nil || affected == 0 { return ErrStaleRun } return tx.Commit() } func (s *Store) LatestFailedJob(ctx context.Context) (*domain.Job, error) { return s.latest(ctx, `WHERE status=?`, domain.JobFailed) } func (s *Store) LatestJob(ctx context.Context) (*domain.Job, error) { return s.latest(ctx, ``, nil) } func (s *Store) latest(ctx context.Context, where string, arg any) (*domain.Job, error) { query := `SELECT id,repo,pr_number,head_sha,trigger_comment_id,COALESCE(trigger_comment_body,''),command,COALESCE(command_args,''),requested_by,status,COALESCE(last_error,''),COALESCE(result_json,''),created_at,updated_at,started_at,finished_at FROM review_jobs ` + where + ` ORDER BY created_at DESC,id DESC LIMIT 1` var row *sql.Row if arg == nil { row = s.db.QueryRowContext(ctx, query) } else { row = s.db.QueryRowContext(ctx, query, arg) } job, err := scanJob(row) if errors.Is(err, sql.ErrNoRows) { return nil, nil } return &job, err } func (s *Store) LatestSuccessfulReview(ctx context.Context, repo string, pr int) (*domain.Job, error) { return s.latestWith(ctx, `WHERE repo=? AND pr_number=? AND command IN ('review','rerun') AND status=?`, repo, pr, domain.JobSucceeded) } func (s *Store) latestWith(ctx context.Context, where string, args ...any) (*domain.Job, error) { query := `SELECT id,repo,pr_number,head_sha,trigger_comment_id,COALESCE(trigger_comment_body,''),command,COALESCE(command_args,''),requested_by,status,COALESCE(last_error,''),COALESCE(result_json,''),created_at,updated_at,started_at,finished_at FROM review_jobs ` + where + ` ORDER BY id DESC LIMIT 1` job, err := scanJob(s.db.QueryRowContext(ctx, query, args...)) if errors.Is(err, sql.ErrNoRows) { return nil, nil } return &job, err } func (s *Store) PendingCount(ctx context.Context, repo string, pr int) (int, error) { var n int err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM review_jobs WHERE repo=? AND pr_number=? AND status IN (?,?)`, repo, pr, domain.JobQueued, domain.JobRunning).Scan(&n) return n, err } func (s *Store) UpsertBotComment(ctx context.Context, repo string, pr int, marker string, commentID int64, sha string) error { if s.dialect == "sqlite" { _, err := s.db.ExecContext(ctx, `INSERT INTO bot_comments(repo,pr_number,head_sha,gitea_comment_id,marker) VALUES(?,?,?,?,?) ON CONFLICT(repo,pr_number,marker) DO UPDATE SET head_sha=excluded.head_sha,gitea_comment_id=excluded.gitea_comment_id,updated_at=CURRENT_TIMESTAMP`, repo, pr, sha, commentID, marker) return err } _, err := s.db.ExecContext(ctx, `INSERT INTO bot_comments(repo,pr_number,head_sha,gitea_comment_id,marker) VALUES(?,?,?,?,?) ON DUPLICATE KEY UPDATE head_sha=VALUES(head_sha),gitea_comment_id=VALUES(gitea_comment_id),updated_at=CURRENT_TIMESTAMP(6)`, repo, pr, sha, commentID, marker) return err } func (s *Store) BotCommentID(ctx context.Context, repo string, pr int, marker string) (int64, error) { var id int64 err := s.db.QueryRowContext(ctx, `SELECT gitea_comment_id FROM bot_comments WHERE repo=? AND pr_number=? AND marker=?`, repo, pr, marker).Scan(&id) if errors.Is(err, sql.ErrNoRows) { return 0, nil } return id, err } func (s *Store) getJob(ctx context.Context, id int64) (domain.Job, error) { return scanJob(s.db.QueryRowContext(ctx, `SELECT id,repo,pr_number,head_sha,trigger_comment_id,COALESCE(trigger_comment_body,''),command,COALESCE(command_args,''),requested_by,status,COALESCE(last_error,''),COALESCE(result_json,''),created_at,updated_at,started_at,finished_at FROM review_jobs WHERE id=?`, id)) } func scanJob(scanner interface{ Scan(...any) error }) (domain.Job, error) { var j domain.Job var status string var result, body, args, last sql.NullString var started, finished sql.NullTime if err := scanner.Scan(&j.ID, &j.Repo, &j.PRNumber, &j.HeadSHA, &j.TriggerCommentID, &body, &j.Command, &args, &j.RequestedBy, &status, &last, &result, &j.CreatedAt, &j.UpdatedAt, &started, &finished); err != nil { return domain.Job{}, err } j.TriggerCommentBody = body.String j.CommandArgs = args.String j.LastError = last.String j.ResultJSON = []byte(result.String) j.Status = domain.JobStatus(status) if started.Valid { j.StartedAt = &started.Time } if finished.Valid { j.FinishedAt = &finished.Time } return j, nil } func isConstraint(err error) bool { text := strings.ToLower(err.Error()) return strings.Contains(text, "unique") || strings.Contains(text, "duplicate") || strings.Contains(text, "constraint") } func nullable(v string) any { if v == "" { return nil } return v } func nullableBytes(v []byte) any { if len(v) == 0 { return nil } return string(v) } var _ store.Store = (*Store)(nil)