Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877package db
import ( "context" "database/sql" "errors" "fmt" "log/slog" "strings" "time"
"tangled.org/core/log" "tangled.org/core/sqlite")
var ( ErrNotFound = errors.New("not found") ErrConflict = errors.New("idempotency conflict") ErrInvalidTransition = errors.New("invalid status transition"))
type InvalidTransitionError struct { Status JobStatus}
func (e *InvalidTransitionError) Error() string { return fmt.Sprintf("%s: cannot retry job with status %q (must be failed or authorization_required)", ErrInvalidTransition, e.Status)}
func (e *InvalidTransitionError) Is(target error) bool { return target == ErrInvalidTransition }
type DB struct { *sql.DB logger *slog.Logger
MaxActivePerOwner int}
func Make(ctx context.Context, dbPath string) (*DB, error) { logger := log.SubLogger(log.FromContext(ctx), "db")
db, err := sqlite.Open(dbPath) if err != nil { return nil, fmt.Errorf("opening sqlite database: %w", err) }
conn, err := db.Conn(ctx) if err != nil { return nil, fmt.Errorf("getting sqlite connection: %w", err) } defer conn.Close()
if _, err := conn.ExecContext(ctx, schema); err != nil { return nil, fmt.Errorf("applying schema: %w", err) } if err := addColumnIfMissing(ctx, conn, "jobs", "capability_token", "text"); err != nil { return nil, fmt.Errorf("migrating jobs capability token: %w", err) } if err := addColumnIfMissing(ctx, conn, "jobs", "description", "text not null default ''"); err != nil { return nil, fmt.Errorf("migrating jobs description: %w", err) } if _, err := conn.ExecContext(ctx, "create unique index if not exists idx_jobs_capability_token on jobs(capability_token) where capability_token is not null"); err != nil { return nil, fmt.Errorf("indexing jobs capability token: %w", err) }
return &DB{DB: db, logger: logger}, nil}
const schema = `create table if not exists batches ( id text primary key, owner_did text not null, request_id text not null, request_digest text not null, encrypted_token text, credential_expires_at text, created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), unique(owner_did, request_id));
create index if not exists idx_batches_owner_created on batches(owner_did, created_at desc);create index if not exists idx_batches_cred_exp on batches(credential_expires_at);
create table if not exists jobs ( id integer primary key autoincrement, batch_id text not null references batches(id) on delete cascade, repo_did text not null, name text not null, knot_did text not null, source_url text not null, description text not null default '', private integer not null default 0, status text not null default 'queued', attempts integer not null default 0, next_attempt_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), error text, capability_token text, created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')));
create index if not exists idx_jobs_batch_id on jobs(batch_id);create index if not exists idx_jobs_status_due on jobs(status, next_attempt_at);create table if not exists oauth_sessions ( owner_did text primary key, session_id text not null, encrypted_data text not null, updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')));
create table if not exists oauth_requests ( state text primary key, encrypted_data text not null, created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')));`
func addColumnIfMissing(ctx context.Context, conn *sql.Conn, table, column, definition string) error { rows, err := conn.QueryContext(ctx, "pragma table_info("+table+")") if err != nil { return err } defer rows.Close() for rows.Next() { var cid int var name, typ string var notNull, pk int var defaultValue sql.NullString if err := rows.Scan(&cid, &name, &typ, ¬Null, &defaultValue, &pk); err != nil { return err } if name == column { return nil } } if err := rows.Err(); err != nil { return err } _, err = conn.ExecContext(ctx, "alter table "+table+" add column "+column+" "+definition) return err}
type CreateBatchInput struct { ID string OwnerDid string RequestID string RequestDigest string EncryptedToken *string Jobs []CreateJobInput}
type CreateJobInput struct { Name string KnotDid string SourceURL string Description string Private bool RepoDid string}
func (d *DB) CreateBatch(ctx context.Context, in CreateBatchInput) (*Batch, []Job, bool, error) { tx, err := d.BeginTx(ctx, nil) if err != nil { return nil, nil, false, err } defer tx.Rollback()
var existing Batch var existingToken sql.NullString var credExpStr sql.NullString var createdAtStr, updatedAtStr string err = tx.QueryRowContext(ctx, ` select id, owner_did, request_id, request_digest, encrypted_token, credential_expires_at, created_at, updated_at from batches where owner_did = ? and request_id = ? `, in.OwnerDid, in.RequestID).Scan( &existing.ID, &existing.OwnerDid, &existing.RequestID, &existing.RequestDigest, &existingToken, &credExpStr, &createdAtStr, &updatedAtStr, )
if err == nil { if existing.RequestDigest != in.RequestDigest { return nil, nil, false, ErrConflict } if existingToken.Valid { existing.EncryptedToken = &existingToken.String } if credExpStr.Valid { t, _ := time.Parse(time.RFC3339, credExpStr.String) existing.CredentialExpiresAt = &t } existing.CreatedAt, _ = time.Parse(time.RFC3339, createdAtStr) existing.UpdatedAt, _ = time.Parse(time.RFC3339, updatedAtStr)
jobs, err := jobsForBatch(ctx, tx, existing.ID) if err != nil { return nil, nil, false, err } return &existing, jobs, true, nil } else if !errors.Is(err, sql.ErrNoRows) { return nil, nil, false, fmt.Errorf("querying existing batch: %w", err) }
now := time.Now().UTC() nowStr := now.Format(time.RFC3339)
var tokenArg any var credExpArg any var credExpiresAt *time.Time if in.EncryptedToken != nil { tokenArg = *in.EncryptedToken exp := now.Add(1 * time.Hour) credExpiresAt = &exp credExpArg = exp.Format(time.RFC3339) }
_, err = tx.ExecContext(ctx, ` insert into batches (id, owner_did, request_id, request_digest, encrypted_token, credential_expires_at, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?) `, in.ID, in.OwnerDid, in.RequestID, in.RequestDigest, tokenArg, credExpArg, nowStr, nowStr) if err != nil { return nil, nil, false, fmt.Errorf("inserting batch: %w", err) }
createdJobs := make([]Job, 0, len(in.Jobs)) for _, j := range in.Jobs { privInt := 0 if j.Private { privInt = 1 } res, err := tx.ExecContext(ctx, ` insert into jobs (batch_id, repo_did, name, knot_did, source_url, description, private, status, attempts, next_attempt_at, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, 'queued', 0, ?, ?, ?) `, in.ID, j.RepoDid, j.Name, j.KnotDid, j.SourceURL, j.Description, privInt, nowStr, nowStr, nowStr) if err != nil { return nil, nil, false, fmt.Errorf("inserting job: %w", err) } jobID, err := res.LastInsertId() if err != nil { return nil, nil, false, fmt.Errorf("getting last insert id: %w", err) } createdJobs = append(createdJobs, Job{ ID: jobID, BatchID: in.ID, RepoDid: j.RepoDid, Name: j.Name, KnotDid: j.KnotDid, SourceURL: j.SourceURL, Description: j.Description, Private: j.Private, Status: StatusQueued, Attempts: 0, NextAttemptAt: now, CreatedAt: now, UpdatedAt: now, }) }
if err := tx.Commit(); err != nil { return nil, nil, false, fmt.Errorf("committing transaction: %w", err) }
batch := &Batch{ ID: in.ID, OwnerDid: in.OwnerDid, RequestID: in.RequestID, RequestDigest: in.RequestDigest, EncryptedToken: in.EncryptedToken, CredentialExpiresAt: credExpiresAt, CreatedAt: now, UpdatedAt: now, }
return batch, createdJobs, false, nil}
type querier interface { QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)}
func mapSlice[T, U any](items []T, f func(T) U) []U { out := make([]U, len(items)) for i, it := range items { out[i] = f(it) } return out}
func jobsForBatches(ctx context.Context, q querier, batchIDs []string) (map[string][]Job, error) { if len(batchIDs) == 0 { return map[string][]Job{}, nil } placeholders := strings.TrimSuffix(strings.Repeat("?,", len(batchIDs)), ",") args := mapSlice(batchIDs, func(id string) any { return id })
rows, err := q.QueryContext(ctx, ` select id, batch_id, repo_did, name, knot_did, source_url, description, private, capability_token, status, attempts, next_attempt_at, error, created_at, updated_at from jobs where batch_id in (`+placeholders+`) order by batch_id asc, id asc `, args...) if err != nil { return nil, fmt.Errorf("querying jobs: %w", err) } defer rows.Close()
jobs := make(map[string][]Job, len(batchIDs)) for rows.Next() { var j Job var privInt int var errStr, capabilityToken sql.NullString var createdStr, updatedStr, nextAttStr string var statusStr string if err := rows.Scan( &j.ID, &j.BatchID, &j.RepoDid, &j.Name, &j.KnotDid, &j.SourceURL, &j.Description, &privInt, &capabilityToken, &statusStr, &j.Attempts, &nextAttStr, &errStr, &createdStr, &updatedStr, ); err != nil { return nil, fmt.Errorf("scanning job: %w", err) } j.Private = privInt != 0 if capabilityToken.Valid { j.CapabilityToken = &capabilityToken.String } j.Status = JobStatus(statusStr) if errStr.Valid { j.Error = &errStr.String } j.NextAttemptAt, _ = time.Parse(time.RFC3339, nextAttStr) j.CreatedAt, _ = time.Parse(time.RFC3339, createdStr) j.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr) jobs[j.BatchID] = append(jobs[j.BatchID], j) } return jobs, rows.Err()}
func jobsForBatch(ctx context.Context, q querier, batchID string) ([]Job, error) { byBatch, err := jobsForBatches(ctx, q, []string{batchID}) if err != nil { return nil, err } return byBatch[batchID], nil}
func (d *DB) GetBatch(ctx context.Context, batchID string) (*Batch, []Job, error) { var b Batch var encToken sql.NullString var credExpStr sql.NullString var createdStr, updatedStr string
err := d.QueryRowContext(ctx, ` select id, owner_did, request_id, request_digest, encrypted_token, credential_expires_at, created_at, updated_at from batches where id = ? `, batchID).Scan( &b.ID, &b.OwnerDid, &b.RequestID, &b.RequestDigest, &encToken, &credExpStr, &createdStr, &updatedStr, ) if errors.Is(err, sql.ErrNoRows) { return nil, nil, ErrNotFound } if err != nil { return nil, nil, fmt.Errorf("querying batch: %w", err) }
if encToken.Valid { b.EncryptedToken = &encToken.String } if credExpStr.Valid { t, _ := time.Parse(time.RFC3339, credExpStr.String) b.CredentialExpiresAt = &t } b.CreatedAt, _ = time.Parse(time.RFC3339, createdStr) b.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr)
jobs, err := jobsForBatch(ctx, d, batchID) if err != nil { return nil, nil, err }
return &b, jobs, nil}
func (d *DB) GetRecentTasksForOwner(ctx context.Context, ownerDid string, limit int) ([]Task, error) { if limit <= 0 { limit = 20 }
rows, err := d.QueryContext(ctx, ` select id, owner_did, created_at, updated_at from batches where owner_did = ? order by created_at desc, id desc limit ? `, ownerDid, limit) if err != nil { return nil, fmt.Errorf("querying recent tasks: %w", err) } defer rows.Close()
tasks := []Task{} for rows.Next() { var t Task var createdStr, updatedStr string if err := rows.Scan(&t.ID, &t.OwnerDid, &createdStr, &updatedStr); err != nil { return nil, fmt.Errorf("scanning task: %w", err) } t.CreatedAt, _ = time.Parse(time.RFC3339, createdStr) t.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr) tasks = append(tasks, t) } if err := rows.Err(); err != nil { return nil, err }
ids := mapSlice(tasks, func(t Task) string { return t.ID }) byBatch, err := jobsForBatches(ctx, d, ids) if err != nil { return nil, err } for i := range tasks { tasks[i].Jobs = byBatch[tasks[i].ID] }
return tasks, nil}
func (d *DB) GetJob(ctx context.Context, jobID int64) (*Job, *Batch, error) { var j Job var b Batch var privInt int var errStr sql.NullString var jCreatedStr, jUpdatedStr, nextStr string var bCreatedStr, bUpdatedStr string var encToken sql.NullString var capabilityToken sql.NullString var credExpStr sql.NullString var statusStr string
err := d.QueryRowContext(ctx, ` select j.id, j.batch_id, j.repo_did, j.name, j.knot_did, j.source_url, j.description, j.private, j.capability_token, j.status, j.attempts, j.next_attempt_at, j.error, j.created_at, j.updated_at, b.id, b.owner_did, b.request_id, b.request_digest, b.encrypted_token, b.credential_expires_at, b.created_at, b.updated_at from jobs j join batches b on j.batch_id = b.id where j.id = ? `, jobID).Scan( &j.ID, &j.BatchID, &j.RepoDid, &j.Name, &j.KnotDid, &j.SourceURL, &j.Description, &privInt, &capabilityToken, &statusStr, &j.Attempts, &nextStr, &errStr, &jCreatedStr, &jUpdatedStr, &b.ID, &b.OwnerDid, &b.RequestID, &b.RequestDigest, &encToken, &credExpStr, &bCreatedStr, &bUpdatedStr, ) if errors.Is(err, sql.ErrNoRows) { return nil, nil, ErrNotFound } if err != nil { return nil, nil, fmt.Errorf("querying job: %w", err) }
j.Private = privInt != 0 if capabilityToken.Valid { j.CapabilityToken = &capabilityToken.String } j.Status = JobStatus(statusStr) if errStr.Valid { j.Error = &errStr.String } j.NextAttemptAt, _ = time.Parse(time.RFC3339, nextStr) j.CreatedAt, _ = time.Parse(time.RFC3339, jCreatedStr) j.UpdatedAt, _ = time.Parse(time.RFC3339, jUpdatedStr)
if encToken.Valid { b.EncryptedToken = &encToken.String } if credExpStr.Valid { t, _ := time.Parse(time.RFC3339, credExpStr.String) b.CredentialExpiresAt = &t } b.CreatedAt, _ = time.Parse(time.RFC3339, bCreatedStr) b.UpdatedAt, _ = time.Parse(time.RFC3339, bUpdatedStr)
return &j, &b, nil}
func (d *DB) ClaimNextQueuedJob(ctx context.Context) (*Job, *Batch, error) { tx, err := d.BeginTx(ctx, nil) if err != nil { return nil, nil, err } defer tx.Rollback()
now := time.Now().UTC() nowStr := now.Format(time.RFC3339)
// fair-share across owners so one queue cannot starve others query := ` with owner_active as ( select b.owner_did, count(*) as active from jobs j join batches b on j.batch_id = b.id where j.status in ('cloning', 'importing') group by b.owner_did ) select j.id from jobs j join batches b on j.batch_id = b.id left join owner_active a on a.owner_did = b.owner_did where j.status = 'queued' and j.next_attempt_at <= ? ` args := []any{nowStr} if d.MaxActivePerOwner > 0 { query += " and coalesce(a.active, 0) < ?" args = append(args, d.MaxActivePerOwner) } query += " order by coalesce(a.active, 0) asc, j.next_attempt_at asc, j.id asc limit 1"
var jobID int64 err = tx.QueryRowContext(ctx, query, args...).Scan(&jobID) if errors.Is(err, sql.ErrNoRows) { return nil, nil, nil } if err != nil { return nil, nil, fmt.Errorf("finding queued job: %w", err) }
res, err := tx.ExecContext(ctx, ` update jobs set status = 'cloning', attempts = attempts + 1, updated_at = ? where id = ? and status = 'queued' `, nowStr, jobID) if err != nil { return nil, nil, fmt.Errorf("claiming job: %w", err) } rowsAffected, _ := res.RowsAffected() if rowsAffected == 0 { return nil, nil, nil }
if err := tx.Commit(); err != nil { return nil, nil, fmt.Errorf("committing claim: %w", err) }
return d.GetJob(ctx, jobID)}
func (d *DB) ReleaseJob(ctx context.Context, jobID int64, errMsg *string) error { nowStr := time.Now().UTC().Format(time.RFC3339) var errVal any if errMsg != nil { errVal = *errMsg }
res, err := d.ExecContext(ctx, ` update jobs set status = 'queued', attempts = max(attempts - 1, 0), next_attempt_at = ?, error = ?, updated_at = ? where id = ? `, nowStr, errVal, nowStr, jobID) if err != nil { return fmt.Errorf("releasing job: %w", err) } if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } return nil}
func (d *DB) UpdateJobStatus(ctx context.Context, jobID int64, status JobStatus, errMsg *string) error { nowStr := time.Now().UTC().Format(time.RFC3339) var errVal any if errMsg != nil { errVal = *errMsg }
res, err := d.ExecContext(ctx, ` update jobs set status = ?, error = ?, updated_at = ? where id = ? `, string(status), errVal, nowStr, jobID) if err != nil { return fmt.Errorf("updating job status: %w", err) } rowsAffected, _ := res.RowsAffected() if rowsAffected == 0 { return ErrNotFound } return nil}
func (d *DB) ScheduleJobRetry(ctx context.Context, jobID int64, backoff time.Duration, errMsg *string) error { now := time.Now().UTC() nextAttemptAt := now.Add(backoff).Format(time.RFC3339) nowStr := now.Format(time.RFC3339)
var errVal any if errMsg != nil { errVal = *errMsg }
// repo_did, capability, and mirror survive retry so worker can resume res, err := d.ExecContext(ctx, ` update jobs set status = 'queued', next_attempt_at = ?, error = ?, updated_at = ? where id = ? `, nextAttemptAt, errVal, nowStr, jobID) if err != nil { return fmt.Errorf("scheduling job retry: %w", err) } rowsAffected, _ := res.RowsAffected() if rowsAffected == 0 { return ErrNotFound } return nil}
func (d *DB) RequeueJob(ctx context.Context, jobID int64, expectedOwnerDid string, newEncryptedToken *string) (*Batch, []Job, error) { tx, err := d.BeginTx(ctx, nil) if err != nil { return nil, nil, err } defer tx.Rollback()
var batchID, ownerDid, currentStatus string var privateInt int err = tx.QueryRowContext(ctx, ` select b.id, b.owner_did, j.status, j.private from jobs j join batches b on j.batch_id = b.id where j.id = ? `, jobID).Scan(&batchID, &ownerDid, ¤tStatus, &privateInt) if errors.Is(err, sql.ErrNoRows) { return nil, nil, ErrNotFound } if err != nil { return nil, nil, fmt.Errorf("finding job: %w", err) }
if expectedOwnerDid != "" && ownerDid != expectedOwnerDid { return nil, nil, ErrNotFound }
status := JobStatus(currentStatus) if !status.CanRetry() { return nil, nil, &InvalidTransitionError{Status: status} }
now := time.Now().UTC() nowStr := now.Format(time.RFC3339)
if newEncryptedToken != nil { newExpStr := now.Add(1 * time.Hour).Format(time.RFC3339) _, err = tx.ExecContext(ctx, ` update batches set encrypted_token = ?, credential_expires_at = ?, updated_at = ? where id = ? `, *newEncryptedToken, newExpStr, nowStr, batchID) if err != nil { return nil, nil, fmt.Errorf("updating batch token: %w", err) } }
// repo_did and capability survive manual retry so worker can resume _, err = tx.ExecContext(ctx, ` update jobs set status = 'queued', attempts = 0, next_attempt_at = ?, error = null, updated_at = ? where id = ? `, nowStr, nowStr, jobID) if err != nil { return nil, nil, fmt.Errorf("requeueing job: %w", err) }
if err := tx.Commit(); err != nil { return nil, nil, fmt.Errorf("committing requeue: %w", err) }
return d.GetBatch(ctx, batchID)}
func (d *DB) SetJobCapability(ctx context.Context, jobID int64, token string) error { if token == "" { return errors.New("empty capability token") } res, err := d.ExecContext(ctx, ` update jobs set capability_token = ?, updated_at = ? where id = ? and status = 'importing' `, token, time.Now().UTC().Format(time.RFC3339), jobID) if err != nil { return fmt.Errorf("publishing job capability: %w", err) } if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } return nil}
func (d *DB) RevokeJobCapability(ctx context.Context, jobID int64) error { _, err := d.ExecContext(ctx, ` update jobs set capability_token = null, updated_at = ? where id = ? `, time.Now().UTC().Format(time.RFC3339), jobID) if err != nil { return fmt.Errorf("revoking job capability: %w", err) } return nil}
func (d *DB) ResolveJobCapability(ctx context.Context, token, name string) (*Job, error) { var jobID int64 // capability resolves on queued so a knot fetch can complete during retry backoff err := d.QueryRowContext(ctx, ` select id from jobs where capability_token = ? and name = ? and status in ('queued', 'cloning', 'importing') `, token, name).Scan(&jobID) if errors.Is(err, sql.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("resolving job capability: %w", err) } job, _, err := d.GetJob(ctx, jobID) return job, err}
func (d *DB) SetJobRepoDid(ctx context.Context, jobID int64, repoDid string) error { res, err := d.ExecContext(ctx, ` update jobs set repo_did = ?, updated_at = ? where id = ? `, repoDid, time.Now().UTC().Format(time.RFC3339), jobID) if err != nil { return fmt.Errorf("recording created repo did: %w", err) } if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } return nil}
type EarlierRepoAttempt struct { RepoDid string SourceURL string}
func (d *DB) EarlierRepoAttemptsFor(ctx context.Context, ownerDid, knotDid, name string) ([]EarlierRepoAttempt, error) { rows, err := d.QueryContext(ctx, ` select j.repo_did, j.source_url from jobs j join batches b on j.batch_id = b.id where b.owner_did = ? and j.knot_did = ? and j.name = ? and j.repo_did != '' order by j.id desc `, ownerDid, knotDid, name) if err != nil { return nil, fmt.Errorf("finding earlier repo attempts for %s/%s: %w", ownerDid, name, err) } defer rows.Close()
var attempts []EarlierRepoAttempt for rows.Next() { var attempt EarlierRepoAttempt if err := rows.Scan(&attempt.RepoDid, &attempt.SourceURL); err != nil { return nil, fmt.Errorf("scanning an earlier repo attempt for %s/%s: %w", ownerDid, name, err) } attempts = append(attempts, attempt) } return attempts, rows.Err()}
// Mirrors the request digest normalization.func SameSource(a, b string) bool { normalize := func(raw string) string { return strings.TrimSuffix(strings.ToLower(strings.TrimSpace(raw)), ".git") } return normalize(a) != "" && normalize(a) == normalize(b)}
func (d *DB) ResetInflightJobsToQueued(ctx context.Context) (int64, error) { nowStr := time.Now().UTC().Format(time.RFC3339) // repo_did survives crash recovery so the worker can resume from the knot res, err := d.ExecContext(ctx, ` update jobs set status = 'queued', next_attempt_at = ?, capability_token = null, updated_at = ? where status in ('cloning', 'importing') `, nowStr, nowStr) if err != nil { return 0, fmt.Errorf("resetting inflight jobs: %w", err) } return res.RowsAffected()}
func (d *DB) CheckAndScrubBatchToken(ctx context.Context, batchID string) (bool, error) { tx, err := d.BeginTx(ctx, nil) if err != nil { return false, err } defer tx.Rollback()
var nonTerminalCount int err = tx.QueryRowContext(ctx, ` select count(*) from jobs where batch_id = ? and status not in ('completed', 'failed', 'authorization_required') `, batchID).Scan(&nonTerminalCount) if err != nil { return false, fmt.Errorf("counting non-terminal jobs: %w", err) }
if nonTerminalCount == 0 { nowStr := time.Now().UTC().Format(time.RFC3339) _, err = tx.ExecContext(ctx, ` update batches set encrypted_token = null, updated_at = ? where id = ? and encrypted_token is not null `, nowStr, batchID) if err != nil { return false, fmt.Errorf("scrubbing batch token: %w", err) } if err := tx.Commit(); err != nil { return false, fmt.Errorf("committing scrub: %w", err) } return true, nil }
return false, nil}
func (d *DB) ScrubExpiredCredentials(ctx context.Context) (int64, error) { tx, err := d.BeginTx(ctx, nil) if err != nil { return 0, err } defer tx.Rollback()
now := time.Now().UTC() nowStr := now.Format(time.RFC3339)
errMsg := "GitHub credential expired after 1 hour grant window" _, err = tx.ExecContext(ctx, ` update jobs set status = 'authorization_required', error = ?, updated_at = ? where private = 1 and status = 'queued' and batch_id in ( select id from batches where encrypted_token is not null and credential_expires_at <= ? ) `, errMsg, nowStr, nowStr) if err != nil { return 0, fmt.Errorf("marking expired unstarted jobs: %w", err) }
res, err := tx.ExecContext(ctx, ` update batches set encrypted_token = null, updated_at = ? where encrypted_token is not null and credential_expires_at <= ? `, nowStr, nowStr) if err != nil { return 0, fmt.Errorf("scrubbing expired batch tokens: %w", err) }
affected, _ := res.RowsAffected()
if err := tx.Commit(); err != nil { return 0, fmt.Errorf("committing credential expiration: %w", err) }
return affected, nil}