You've already forked agentic-coding-workflow
101 lines
3.8 KiB
Go
101 lines
3.8 KiB
Go
package handlers
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
|
|
"github.com/yan1h/agent-coding-workflow/internal/infra/crypto"
|
|
"github.com/yan1h/agent-coding-workflow/internal/jobs"
|
|
)
|
|
|
|
// ReencryptStore abstracts the DB scan + per-row re-seal so the handler is
|
|
// testable without a live Postgres. ReencryptTable processes one batch of rows
|
|
// for one logical table that are still below activeVersion, re-sealing each
|
|
// blob with enc.EncryptVersion(active) inside a per-row transaction, and returns
|
|
// how many rows it advanced. It returns 0 when no rows remain on an old version
|
|
// (the table is done). Because each row's UPDATE sets both blob+key_version in
|
|
// one tx, the job is idempotent and resumable: re-running after a crash simply
|
|
// finds the rows that did not yet advance.
|
|
type ReencryptStore interface {
|
|
// Tables returns the logical table names this store can re-encrypt.
|
|
Tables() []string
|
|
// ReencryptTable advances up to batch rows of table below activeVersion onto
|
|
// the active version, returning the number of rows updated this call.
|
|
ReencryptTable(ctx context.Context, enc *crypto.KeyedEncryptor, table string, activeVersion, batch int) (int, error)
|
|
// StaleCount reports how many rows of table remain below activeVersion (for
|
|
// the reencrypt-status endpoint and the handler's completion check).
|
|
StaleCount(ctx context.Context, table string, activeVersion int) (int, error)
|
|
}
|
|
|
|
// SecretReencryptPayload is the job payload. ActiveVersion is the target write
|
|
// version (the version crypto_keys is now active on); the handler re-encrypts
|
|
// every stale row onto it. BatchSize bounds per-iteration work; 0 → default.
|
|
type SecretReencryptPayload struct {
|
|
ActiveVersion int `json:"active_version"`
|
|
BatchSize int `json:"batch_size"`
|
|
}
|
|
|
|
// SecretReencryptHandler implements jobs.Handler for TypeSecretReencrypt.
|
|
type SecretReencryptHandler struct {
|
|
store ReencryptStore
|
|
enc *crypto.KeyedEncryptor
|
|
log *slog.Logger
|
|
}
|
|
|
|
// NewSecretReencryptHandler builds the handler.
|
|
func NewSecretReencryptHandler(store ReencryptStore, enc *crypto.KeyedEncryptor, log *slog.Logger) *SecretReencryptHandler {
|
|
if log == nil {
|
|
log = slog.Default()
|
|
}
|
|
return &SecretReencryptHandler{store: store, enc: enc, log: log}
|
|
}
|
|
|
|
// Type reports the job type this handler serves.
|
|
func (h *SecretReencryptHandler) Type() jobs.JobType { return jobs.TypeSecretReencrypt }
|
|
|
|
const defaultReencryptBatch = 100
|
|
|
|
// Handle scans every encrypted-column table and re-seals rows still on an old
|
|
// key version onto the active version, batch by batch, until none remain. It is
|
|
// idempotent (already-advanced rows are skipped by the key_version<active
|
|
// predicate) and resumable (a crash mid-run leaves a partial set; re-running
|
|
// completes the rest). A malformed payload is a permanent error (no retry).
|
|
func (h *SecretReencryptHandler) Handle(ctx context.Context, job jobs.Job) error {
|
|
var p SecretReencryptPayload
|
|
if err := json.Unmarshal(job.Payload, &p); err != nil {
|
|
return jobs.Permanent(fmt.Errorf("secret_reencrypt: bad payload: %w", err))
|
|
}
|
|
active := p.ActiveVersion
|
|
if active <= 0 {
|
|
v, err := h.enc.ActiveVersion(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("secret_reencrypt: resolve active version: %w", err)
|
|
}
|
|
active = v
|
|
}
|
|
batch := p.BatchSize
|
|
if batch <= 0 {
|
|
batch = defaultReencryptBatch
|
|
}
|
|
|
|
for _, table := range h.store.Tables() {
|
|
for {
|
|
if err := ctx.Err(); err != nil {
|
|
return err // canceled/timeout → retry later, resumable
|
|
}
|
|
n, err := h.store.ReencryptTable(ctx, h.enc, table, active, batch)
|
|
if err != nil {
|
|
return fmt.Errorf("secret_reencrypt: table %s: %w", table, err)
|
|
}
|
|
if n == 0 {
|
|
break // table drained
|
|
}
|
|
h.log.Info("secret_reencrypt.batch", "table", table, "rows", n, "active_version", active)
|
|
}
|
|
}
|
|
h.log.Info("secret_reencrypt.done", "active_version", active)
|
|
return nil
|
|
}
|