1package replication23import (4 "context"5 "database/sql"6 "fmt"7 "time"8)910const jobSchema = `CREATE TABLE IF NOT EXISTS replication_jobs (id TEXT PRIMARY KEY, payload TEXT NOT NULL, state TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, phase TEXT NOT NULL DEFAULT 'pending', error TEXT NOT NULL DEFAULT '', stored INTEGER NOT NULL DEFAULT 0, sent INTEGER NOT NULL DEFAULT 0, skipped INTEGER NOT NULL DEFAULT 0, refused INTEGER NOT NULL DEFAULT 0, cursor INTEGER NOT NULL DEFAULT 0, rounds INTEGER NOT NULL DEFAULT 0, started_at INTEGER NOT NULL DEFAULT 0, finished_at INTEGER NOT NULL DEFAULT 0); CREATE TABLE IF NOT EXISTS replication_job_cursors (job_id TEXT NOT NULL, target TEXT NOT NULL, cursor INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(job_id,target))`1112// EnsureJobSchema initializes replication-owned job state for callers that13// import configuration before constructing a replication service.14func EnsureJobSchema(db interface {15 Exec(string, ...any) (sql.Result, error)16}) error {17 _, err := db.Exec(jobSchema)18 return err19}2021type ConfigJob struct {22 ID string23 Payload []byte24}2526// ReplaceJobsTx replaces the durable replication jobs represented by specs in27// a caller-owned transaction. It is used by configuration import so policy,28// config metadata and replication state commit atomically.29func ReplaceJobsTx(ctx context.Context, tx *sql.Tx, specs []ConfigJob) error {30 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_jobs`); err != nil {31 return err32 }33 now := time.Now().Unix()34 for _, spec := range specs {35 if spec.ID == "" {36 return fmt.Errorf("replication: config job id is required")37 }38 if _, err := tx.ExecContext(ctx, `INSERT INTO replication_jobs(id,payload,state,created_at,updated_at) VALUES(?,?, 'pending',?,?)`, spec.ID, string(spec.Payload), now, now); err != nil {39 return err40 }41 }42 return nil43}