590 lines
25 KiB
Go
590 lines
25 KiB
Go
package store
|
|
|
|
import (
|
|
"bytes"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"git.ipao.vip/rogee/go-sip/internal/configread"
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
// Store refuses an incompatible database instead of clearing or
|
|
// replaying data whose owner and message identity cannot be proved safe.
|
|
type Store struct{ db *sql.DB }
|
|
|
|
var (
|
|
ErrCommandConflict = errors.New("command id reused with different body")
|
|
ErrDispatcherIdentityMismatch = errors.New("database belongs to another dispatcher")
|
|
)
|
|
|
|
const schema = `
|
|
CREATE TABLE IF NOT EXISTS dispatcher_state (
|
|
dispatcher_id TEXT PRIMARY KEY,
|
|
discovery_ready INTEGER NOT NULL DEFAULT 0 CHECK(discovery_ready IN (0,1)),
|
|
applied_sip_revision INTEGER NOT NULL DEFAULT 0 CHECK(applied_sip_revision >= 0),
|
|
pending_sip_revision INTEGER NOT NULL DEFAULT 0 CHECK(pending_sip_revision >= 0)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS dispatcher_tasks (
|
|
dispatcher_id TEXT NOT NULL,
|
|
tenant_id INTEGER NOT NULL CHECK(tenant_id > 0),
|
|
task_id TEXT NOT NULL,
|
|
task_revision INTEGER NOT NULL CHECK(task_revision > 0),
|
|
status TEXT NOT NULL CHECK(status IN ('running','paused','stopped')),
|
|
control_state TEXT NOT NULL DEFAULT '' CHECK(control_state IN ('','pausing','paused','stopping','stopped','resuming')),
|
|
present INTEGER NOT NULL DEFAULT 1 CHECK(present IN (0,1)),
|
|
PRIMARY KEY(dispatcher_id,tenant_id,task_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS dispatcher_configs (
|
|
dispatcher_id TEXT NOT NULL,
|
|
tenant_id INTEGER NOT NULL CHECK(tenant_id > 0),
|
|
task_id TEXT NOT NULL,
|
|
task_revision INTEGER NOT NULL CHECK(task_revision > 0),
|
|
sip_revision INTEGER NOT NULL CHECK(sip_revision > 0),
|
|
quota_revision INTEGER NOT NULL CHECK(quota_revision > 0),
|
|
content_sha256 TEXT NOT NULL,
|
|
snapshot_json BLOB NOT NULL,
|
|
PRIMARY KEY(dispatcher_id,tenant_id,task_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS dispatcher_inbox (
|
|
dispatcher_id TEXT NOT NULL,
|
|
event_id TEXT NOT NULL,
|
|
tenant_id INTEGER NOT NULL CHECK(tenant_id > 0),
|
|
task_id TEXT NOT NULL,
|
|
callee TEXT NOT NULL,
|
|
issued_at TEXT NOT NULL,
|
|
status TEXT NOT NULL CHECK(status IN ('pending','dispatching','dispatched','unknown','rejected','finished','suppressed')),
|
|
selected_trunk_id TEXT,
|
|
caller_id TEXT,
|
|
dialed_callee TEXT,
|
|
deadline TEXT,
|
|
snapshot_json BLOB,
|
|
PRIMARY KEY(dispatcher_id,event_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS dispatcher_inbox_occupancy ON dispatcher_inbox(dispatcher_id,tenant_id,task_id,selected_trunk_id,status);
|
|
CREATE TABLE IF NOT EXISTS dispatcher_outbox (
|
|
dispatcher_id TEXT NOT NULL,
|
|
event_id TEXT NOT NULL,
|
|
event_type TEXT NOT NULL,
|
|
routing_key TEXT NOT NULL,
|
|
body BLOB NOT NULL,
|
|
confirmed INTEGER NOT NULL DEFAULT 0 CHECK(confirmed IN (0,1)),
|
|
confirmed_at TEXT,
|
|
PRIMARY KEY(dispatcher_id,event_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS dispatcher_recordings (
|
|
dispatcher_id TEXT NOT NULL,
|
|
source_event_id TEXT NOT NULL,
|
|
upload_id TEXT NOT NULL,
|
|
recording_id TEXT NOT NULL,
|
|
bucket TEXT NOT NULL,
|
|
object_key TEXT NOT NULL,
|
|
checksum_sha256 TEXT NOT NULL,
|
|
size_bytes INTEGER NOT NULL CHECK(size_bytes > 0),
|
|
format TEXT NOT NULL,
|
|
channels INTEGER NOT NULL CHECK(channels > 0),
|
|
sample_rate_hz INTEGER NOT NULL CHECK(sample_rate_hz > 0),
|
|
duration_ms INTEGER NOT NULL CHECK(duration_ms >= 0),
|
|
confirmed_at TEXT,
|
|
PRIMARY KEY(dispatcher_id,source_event_id),
|
|
UNIQUE(dispatcher_id,upload_id),
|
|
UNIQUE(bucket,object_key),
|
|
FOREIGN KEY(dispatcher_id,source_event_id) REFERENCES dispatcher_inbox(dispatcher_id,event_id)
|
|
);`
|
|
|
|
func Open(path string) (_ *Store, err error) {
|
|
if path == "" {
|
|
return nil, errors.New("SQLite path is required")
|
|
}
|
|
db, err := sql.Open("sqlite", path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open current SQLite: %w", err)
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
db.Close()
|
|
}
|
|
}()
|
|
db.SetMaxOpenConns(1)
|
|
rows, err := db.Query(`SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("inspect SQLite schema before writing: %w", err)
|
|
}
|
|
allowed := map[string]bool{"dispatcher_state": true, "dispatcher_tasks": true, "dispatcher_configs": true, "dispatcher_inbox": true, "dispatcher_outbox": true, "dispatcher_recordings": true}
|
|
seen := make(map[string]bool, len(allowed))
|
|
var unexpected []string
|
|
for rows.Next() {
|
|
var name string
|
|
if err := rows.Scan(&name); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
if !allowed[name] {
|
|
unexpected = append(unexpected, name)
|
|
} else {
|
|
seen[name] = true
|
|
}
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("inspect SQLite tables: %w", err)
|
|
}
|
|
if len(unexpected) > 0 {
|
|
return nil, fmt.Errorf("old or unknown SQLite tables %s: preserve database and reconcile manually before admission", strings.Join(unexpected, ","))
|
|
}
|
|
var version int
|
|
if err := db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil {
|
|
return nil, fmt.Errorf("inspect SQLite layout version before writing: %w", err)
|
|
}
|
|
if len(seen) == 0 {
|
|
if version != 0 {
|
|
return nil, fmt.Errorf("unknown SQLite layout version %d without tables: preserve database before admission", version)
|
|
}
|
|
} else {
|
|
if len(seen) != len(allowed) || version != 2 {
|
|
return nil, fmt.Errorf("existing SQLite layout is incomplete or obsolete (tables=%d, version=%d): preserve data before admission", len(seen), version)
|
|
}
|
|
var requiredColumns int
|
|
if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('dispatcher_state') WHERE name IN ('discovery_ready','applied_sip_revision','pending_sip_revision')`).Scan(&requiredColumns); err != nil {
|
|
return nil, fmt.Errorf("inspect durable SIP state layout: %w", err)
|
|
}
|
|
if requiredColumns != 3 {
|
|
return nil, errors.New("existing SQLite lacks durable SIP admission columns: preserve data before admission")
|
|
}
|
|
var recordingColumns int
|
|
if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('dispatcher_recordings') WHERE name IN ('dispatcher_id','source_event_id','upload_id','recording_id','bucket','object_key','checksum_sha256','size_bytes','format','channels','sample_rate_hz','duration_ms','confirmed_at')`).Scan(&recordingColumns); err != nil {
|
|
return nil, fmt.Errorf("inspect original recording target layout: %w", err)
|
|
}
|
|
if recordingColumns != 13 {
|
|
return nil, errors.New("existing SQLite lacks immutable recording target columns: preserve data before admission")
|
|
}
|
|
}
|
|
for _, pragma := range []string{"PRAGMA busy_timeout=5000", "PRAGMA journal_mode=WAL", "PRAGMA synchronous=FULL", "PRAGMA foreign_keys=ON"} {
|
|
if _, err := db.Exec(pragma); err != nil {
|
|
return nil, fmt.Errorf("initialize SQLite %s: %w", pragma, err)
|
|
}
|
|
}
|
|
if len(seen) == 0 {
|
|
tx, err := db.Begin()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.Exec(schema); err != nil {
|
|
return nil, fmt.Errorf("create current SQLite schema: %w", err)
|
|
}
|
|
if _, err := tx.Exec(`PRAGMA user_version=2`); err != nil {
|
|
return nil, fmt.Errorf("set current SQLite layout version: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, fmt.Errorf("commit current SQLite schema: %w", err)
|
|
}
|
|
} else if _, err := db.Exec(schema); err != nil {
|
|
return nil, fmt.Errorf("verify current SQLite schema: %w", err)
|
|
}
|
|
return &Store{db: db}, nil
|
|
}
|
|
|
|
func (s *Store) Close() error { return s.db.Close() }
|
|
|
|
// CloseAdmission is durable, including when startup fails before SIP loading.
|
|
func (s *Store) CloseAdmission(dispatcherID string) error {
|
|
if dispatcherID == "" {
|
|
return errors.New("dispatcher ID is required")
|
|
}
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return fmt.Errorf("begin Dispatcher admission fence: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
var boundID string
|
|
checkErr := tx.QueryRow(`SELECT dispatcher_id FROM dispatcher_state WHERE dispatcher_id<>? LIMIT 1`, dispatcherID).Scan(&boundID)
|
|
if checkErr == nil {
|
|
return fmt.Errorf("%w: stored=%s requested=%s", ErrDispatcherIdentityMismatch, boundID, dispatcherID)
|
|
}
|
|
if !errors.Is(checkErr, sql.ErrNoRows) {
|
|
return fmt.Errorf("read SQLite Dispatcher owner: %w", checkErr)
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO dispatcher_state(dispatcher_id,discovery_ready) VALUES(?,0)
|
|
ON CONFLICT(dispatcher_id) DO UPDATE SET discovery_ready=0`, dispatcherID); err != nil {
|
|
return fmt.Errorf("close task admission: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit Dispatcher admission fence: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ApplyDiscoverySnapshot commits the entire cold-start list in one transaction.
|
|
// It closes admission until the caller has drained the control queue.
|
|
func (s *Store) ApplyDiscoverySnapshot(dispatcherID string, tasks []configread.DiscoveredTask) (err error) {
|
|
if dispatcherID == "" {
|
|
return errors.New("dispatcher ID is required")
|
|
}
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err = tx.Exec(`INSERT INTO dispatcher_state(dispatcher_id,discovery_ready) VALUES(?,0)
|
|
ON CONFLICT(dispatcher_id) DO UPDATE SET discovery_ready=0`, dispatcherID); err != nil {
|
|
return fmt.Errorf("close discovery admission: %w", err)
|
|
}
|
|
if _, err = tx.Exec(`UPDATE dispatcher_tasks SET present=0 WHERE dispatcher_id=?`, dispatcherID); err != nil {
|
|
return fmt.Errorf("invalidate previous discovery list: %w", err)
|
|
}
|
|
if err := applyDiscoveredTasks(tx, dispatcherID, tasks); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
// ApplyDiscoveryPage commits one delta page before its in-memory cursor may
|
|
// advance. It never treats tasks absent from a delta page as retired.
|
|
func (s *Store) ApplyDiscoveryPage(dispatcherID string, tasks []configread.DiscoveredTask) error {
|
|
if dispatcherID == "" || len(tasks) == 0 {
|
|
return errors.New("discovery delta requires a Dispatcher and nonempty page")
|
|
}
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if err := applyDiscoveredTasks(tx, dispatcherID, tasks); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit discovery page: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func applyDiscoveredTasks(tx *sql.Tx, dispatcherID string, tasks []configread.DiscoveredTask) error {
|
|
seen := make(map[string]struct{}, len(tasks))
|
|
for _, task := range tasks {
|
|
if task.TaskID == "" || task.TenantID <= 0 || task.TaskRevision <= 0 || !validTaskStatus(task.Status) {
|
|
return fmt.Errorf("invalid discovered task %q", task.TaskID)
|
|
}
|
|
if _, exists := seen[task.TaskID]; exists {
|
|
return fmt.Errorf("duplicate discovered task %q", task.TaskID)
|
|
}
|
|
seen[task.TaskID] = struct{}{}
|
|
var otherTenant int64
|
|
err := tx.QueryRow(`SELECT tenant_id FROM dispatcher_tasks WHERE dispatcher_id=? AND task_id=? AND tenant_id<>? LIMIT 1`, dispatcherID, task.TaskID, task.TenantID).Scan(&otherTenant)
|
|
if err == nil {
|
|
return fmt.Errorf("task %q was previously assigned to a different tenant", task.TaskID)
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("check task %q tenant ownership: %w", task.TaskID, err)
|
|
}
|
|
var oldRevision int64
|
|
err = tx.QueryRow(`SELECT task_revision FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, dispatcherID, task.TenantID, task.TaskID).Scan(&oldRevision)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("check task %q discovery revision: %w", task.TaskID, err)
|
|
}
|
|
if err == nil && task.TaskRevision < oldRevision {
|
|
return fmt.Errorf("task %q discovery revision regressed", task.TaskID)
|
|
}
|
|
initialControl := ""
|
|
if task.Status == "paused" || task.Status == "stopped" {
|
|
initialControl = task.Status
|
|
}
|
|
if _, err = tx.Exec(`INSERT INTO dispatcher_tasks(dispatcher_id,tenant_id,task_id,task_revision,status,present,control_state)
|
|
VALUES(?,?,?,?,?,1,?) ON CONFLICT(dispatcher_id,tenant_id,task_id)
|
|
DO UPDATE SET task_revision=excluded.task_revision,
|
|
status=CASE WHEN dispatcher_tasks.status='stopped' OR dispatcher_tasks.control_state IN ('stopping','stopped') THEN 'stopped' ELSE excluded.status END,
|
|
present=1,
|
|
control_state=CASE
|
|
WHEN dispatcher_tasks.status='stopped' OR dispatcher_tasks.control_state IN ('stopping','stopped') OR excluded.status='stopped' THEN 'stopped'
|
|
WHEN dispatcher_tasks.control_state<>'' THEN dispatcher_tasks.control_state
|
|
WHEN excluded.status='paused' THEN 'paused'
|
|
ELSE '' END`, dispatcherID, task.TenantID, task.TaskID, task.TaskRevision, task.Status, initialControl); err != nil {
|
|
return fmt.Errorf("persist discovered task %q: %w", task.TaskID, err)
|
|
}
|
|
if task.Status == "stopped" || initialControl == "stopped" {
|
|
if _, err := tx.Exec(`UPDATE dispatcher_inbox SET status='suppressed' WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status='pending'`, dispatcherID, task.TenantID, task.TaskID); err != nil {
|
|
return fmt.Errorf("suppress stopped task %q pending calls: %w", task.TaskID, err)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validTaskStatus(status string) bool {
|
|
return status == "running" || status == "paused" || status == "stopped"
|
|
}
|
|
|
|
type AssignedTask struct {
|
|
TenantID int64
|
|
TaskID string
|
|
TaskRevision int64
|
|
Status string
|
|
ControlState string
|
|
}
|
|
|
|
// ListAssignedTasks is used to attach only the task queues assigned to this
|
|
// Dispatcher. Stopped task queues can still drain old, unaccepted commands.
|
|
func (s *Store) ListAssignedTasks(dispatcherID string) ([]AssignedTask, error) {
|
|
rows, err := s.db.Query(`SELECT tenant_id,task_id,task_revision,status,control_state FROM dispatcher_tasks
|
|
WHERE dispatcher_id=? AND present=1 ORDER BY task_id`, dispatcherID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list assigned task queues: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var assigned []AssignedTask
|
|
for rows.Next() {
|
|
var task AssignedTask
|
|
if err := rows.Scan(&task.TenantID, &task.TaskID, &task.TaskRevision, &task.Status, &task.ControlState); err != nil {
|
|
return nil, fmt.Errorf("read assigned task queue: %w", err)
|
|
}
|
|
assigned = append(assigned, task)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("list assigned task queues: %w", err)
|
|
}
|
|
return assigned, nil
|
|
}
|
|
|
|
// CanAdmit checks persisted discovery and human controls; it does not replace
|
|
// call-time whitelist, schedule, SIP load, quota or authorization checks.
|
|
func (s *Store) CanAdmit(dispatcherID string, tenantID int64, taskID string) (bool, error) {
|
|
var count int
|
|
err := s.db.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks t
|
|
JOIN dispatcher_state ds ON ds.dispatcher_id=t.dispatcher_id
|
|
JOIN dispatcher_configs c ON c.dispatcher_id=t.dispatcher_id AND c.tenant_id=t.tenant_id AND c.task_id=t.task_id AND c.task_revision=t.task_revision
|
|
WHERE t.dispatcher_id=? AND t.tenant_id=? AND t.task_id=? AND t.present=1
|
|
AND t.status='running' AND t.control_state='' AND ds.discovery_ready=1`, dispatcherID, tenantID, taskID).Scan(&count)
|
|
if err != nil {
|
|
return false, fmt.Errorf("check durable task admission: %w", err)
|
|
}
|
|
return count == 1, nil
|
|
}
|
|
|
|
// SaveSnapshot refuses an immutable task revision with different content. Only
|
|
// the providers actually referenced by this task are included in its digest.
|
|
func (s *Store) SaveSnapshot(snapshot configread.Snapshot) error {
|
|
task := snapshot.Task
|
|
if task.DispatcherID == "" || task.TenantID <= 0 || task.TaskID == "" || task.TaskRevision <= 0 || snapshot.SIP.DispatcherID != task.DispatcherID || snapshot.Quota.DispatcherID != task.DispatcherID || snapshot.Quota.TenantID != task.TenantID || snapshot.SIP.Revision <= 0 || snapshot.Quota.QuotaRevision <= 0 || !json.Valid(task.Raw) {
|
|
return errors.New("invalid current task snapshot identity, revision, or body")
|
|
}
|
|
providers := make(map[string]configread.Provider)
|
|
for _, ref := range []string{task.Agent.ASR.ProviderRef, task.Agent.LLM.ProviderRef, task.Agent.TTS.ProviderRef} {
|
|
if ref == "" {
|
|
continue
|
|
}
|
|
provider, ok := snapshot.Providers[ref]
|
|
if !ok || !provider.Enabled || provider.Credential == "" {
|
|
return fmt.Errorf("task provider %q is unavailable", ref)
|
|
}
|
|
providers[ref] = provider
|
|
}
|
|
var canonicalTask any
|
|
decoder := json.NewDecoder(bytes.NewReader(task.Raw))
|
|
decoder.UseNumber()
|
|
if err := decoder.Decode(&canonicalTask); err != nil {
|
|
return fmt.Errorf("decode immutable task: %w", err)
|
|
}
|
|
binding, err := json.Marshal(struct {
|
|
Task any `json:"task"`
|
|
Providers map[string]configread.Provider `json:"providers"`
|
|
}{canonicalTask, providers})
|
|
if err != nil {
|
|
return fmt.Errorf("encode immutable task binding: %w", err)
|
|
}
|
|
full, err := json.Marshal(struct {
|
|
Task any `json:"task"`
|
|
Providers map[string]configread.Provider `json:"providers"`
|
|
SIP configread.SIP `json:"sip"`
|
|
Quota configread.Quota `json:"quota"`
|
|
}{canonicalTask, providers, snapshot.SIP, snapshot.Quota})
|
|
if err != nil {
|
|
return fmt.Errorf("encode task execution snapshot: %w", err)
|
|
}
|
|
sum := sha256.Sum256(binding)
|
|
digest := hex.EncodeToString(sum[:])
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if err := validateGlobalRevisions(tx, snapshot); err != nil {
|
|
return err
|
|
}
|
|
var oldRevision int64
|
|
var oldDigest string
|
|
err = tx.QueryRow(`SELECT task_revision,content_sha256 FROM dispatcher_configs WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, task.DispatcherID, task.TenantID, task.TaskID).Scan(&oldRevision, &oldDigest)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("read immutable snapshot version: %w", err)
|
|
}
|
|
if err == nil {
|
|
if oldRevision > task.TaskRevision {
|
|
return fmt.Errorf("task %q revision regressed", task.TaskID)
|
|
}
|
|
if oldRevision == task.TaskRevision && oldDigest != digest {
|
|
return fmt.Errorf("task %q revision %d changed content", task.TaskID, task.TaskRevision)
|
|
}
|
|
if oldRevision == task.TaskRevision {
|
|
if _, err := tx.Exec(`UPDATE dispatcher_configs SET sip_revision=?,quota_revision=?,snapshot_json=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, snapshot.SIP.Revision, snapshot.Quota.QuotaRevision, full, task.DispatcherID, task.TenantID, task.TaskID); err != nil {
|
|
return fmt.Errorf("refresh global SIP/quota for task %q: %w", task.TaskID, err)
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
}
|
|
if _, err = tx.Exec(`INSERT INTO dispatcher_configs(dispatcher_id,tenant_id,task_id,task_revision,sip_revision,quota_revision,content_sha256,snapshot_json)
|
|
VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(dispatcher_id,tenant_id,task_id) DO UPDATE SET
|
|
task_revision=excluded.task_revision,sip_revision=excluded.sip_revision,quota_revision=excluded.quota_revision,
|
|
content_sha256=excluded.content_sha256,snapshot_json=excluded.snapshot_json`, task.DispatcherID, task.TenantID, task.TaskID, task.TaskRevision, snapshot.SIP.Revision, snapshot.Quota.QuotaRevision, digest, full); err != nil {
|
|
return fmt.Errorf("persist task snapshot %q: %w", task.TaskID, err)
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
// validateGlobalRevisions prevents two task bindings from silently
|
|
// disagreeing about the same approved Dispatcher SIP or tenant quota revision.
|
|
// Older revisions also cannot replace a newer binding across tasks.
|
|
func validateGlobalRevisions(tx *sql.Tx, snapshot configread.Snapshot) error {
|
|
newSIP, err := json.Marshal(snapshot.SIP)
|
|
if err != nil {
|
|
return fmt.Errorf("encode approved SIP revision: %w", err)
|
|
}
|
|
newQuota, err := json.Marshal(snapshot.Quota)
|
|
if err != nil {
|
|
return fmt.Errorf("encode approved quota revision: %w", err)
|
|
}
|
|
rows, err := tx.Query(`SELECT tenant_id,task_id,sip_revision,quota_revision,snapshot_json FROM dispatcher_configs WHERE dispatcher_id=?`, snapshot.Task.DispatcherID)
|
|
if err != nil {
|
|
return fmt.Errorf("inspect current global revision bindings: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var tenantID, sipRevision, quotaRevision int64
|
|
var taskID string
|
|
var raw []byte
|
|
if err := rows.Scan(&tenantID, &taskID, &sipRevision, "aRevision, &raw); err != nil {
|
|
return fmt.Errorf("read global revision binding: %w", err)
|
|
}
|
|
var old struct {
|
|
SIP configread.SIP `json:"sip"`
|
|
Quota configread.Quota `json:"quota"`
|
|
}
|
|
if err := json.Unmarshal(raw, &old); err != nil {
|
|
return fmt.Errorf("decode global revision binding for task %q: %w", taskID, err)
|
|
}
|
|
if old.SIP.Revision != sipRevision || old.Quota.QuotaRevision != quotaRevision || old.Quota.TenantID != tenantID {
|
|
return fmt.Errorf("stored global revision binding for task %q is inconsistent", taskID)
|
|
}
|
|
if sipRevision > snapshot.SIP.Revision {
|
|
return fmt.Errorf("SIP revision regressed below task %q", taskID)
|
|
}
|
|
if sipRevision == snapshot.SIP.Revision {
|
|
previous, err := json.Marshal(old.SIP)
|
|
if err != nil {
|
|
return fmt.Errorf("encode existing SIP revision: %w", err)
|
|
}
|
|
if !bytes.Equal(previous, newSIP) {
|
|
return fmt.Errorf("SIP revision %d changed approved content across task %q", sipRevision, taskID)
|
|
}
|
|
}
|
|
if tenantID != snapshot.Task.TenantID {
|
|
continue
|
|
}
|
|
if quotaRevision > snapshot.Quota.QuotaRevision {
|
|
return fmt.Errorf("tenant quota revision regressed below task %q", taskID)
|
|
}
|
|
if quotaRevision == snapshot.Quota.QuotaRevision {
|
|
previous, err := json.Marshal(old.Quota)
|
|
if err != nil {
|
|
return fmt.Errorf("encode existing tenant quota revision: %w", err)
|
|
}
|
|
if !bytes.Equal(previous, newQuota) {
|
|
return fmt.Errorf("tenant quota revision %d changed approved content across task %q", quotaRevision, taskID)
|
|
}
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return fmt.Errorf("read global revision bindings: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ReadSnapshot returns only the current durable binding. Corrupt or mismatched
|
|
// data is an error, never a signal to fetch an old contract instead.
|
|
func (s *Store) ReadSnapshot(dispatcherID string, tenantID int64, taskID string) (configread.Snapshot, error) {
|
|
var body []byte
|
|
var taskRevision, sipRevision, quotaRevision int64
|
|
err := s.db.QueryRow(`SELECT task_revision,sip_revision,quota_revision,snapshot_json FROM dispatcher_configs WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, dispatcherID, tenantID, taskID).Scan(&taskRevision, &sipRevision, "aRevision, &body)
|
|
if err != nil {
|
|
return configread.Snapshot{}, fmt.Errorf("load task execution snapshot: %w", err)
|
|
}
|
|
var decoded struct {
|
|
Task json.RawMessage `json:"task"`
|
|
Providers map[string]configread.Provider `json:"providers"`
|
|
SIP configread.SIP `json:"sip"`
|
|
Quota configread.Quota `json:"quota"`
|
|
}
|
|
if err := json.Unmarshal(body, &decoded); err != nil {
|
|
return configread.Snapshot{}, fmt.Errorf("decode task execution snapshot: %w", err)
|
|
}
|
|
var task configread.Task
|
|
if err := json.Unmarshal(decoded.Task, &task); err != nil {
|
|
return configread.Snapshot{}, fmt.Errorf("decode stored task configuration: %w", err)
|
|
}
|
|
if task.DispatcherID != dispatcherID || task.TenantID != tenantID || task.TaskID != taskID || task.TaskRevision != taskRevision || decoded.SIP.DispatcherID != dispatcherID || decoded.SIP.Revision != sipRevision || decoded.Quota.DispatcherID != dispatcherID || decoded.Quota.TenantID != tenantID || decoded.Quota.QuotaRevision != quotaRevision {
|
|
return configread.Snapshot{}, errors.New("persisted task snapshot identity or revision mismatch")
|
|
}
|
|
return configread.Snapshot{Task: task, Providers: decoded.Providers, SIP: decoded.SIP, Quota: decoded.Quota}, nil
|
|
}
|
|
|
|
func (s *Store) ApplyControl(dispatcherID string, tenantID int64, taskID, action string) error {
|
|
if dispatcherID == "" || tenantID <= 0 || taskID == "" {
|
|
return errors.New("invalid task control identity")
|
|
}
|
|
var newState string
|
|
var predicate string
|
|
switch action {
|
|
case "pause":
|
|
newState, predicate = "paused", "control_state <> 'stopped'"
|
|
case "stop":
|
|
newState, predicate = "stopped", "1=1"
|
|
case "resume":
|
|
newState, predicate = "", `control_state='paused' AND status='running' AND EXISTS
|
|
(SELECT 1 FROM dispatcher_configs c WHERE c.dispatcher_id=dispatcher_tasks.dispatcher_id
|
|
AND c.tenant_id=dispatcher_tasks.tenant_id AND c.task_id=dispatcher_tasks.task_id
|
|
AND c.task_revision=dispatcher_tasks.task_revision)`
|
|
default:
|
|
return fmt.Errorf("unknown task control action %q", action)
|
|
}
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
query := `UPDATE dispatcher_tasks SET control_state=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND present=1 AND ` + predicate
|
|
result, err := tx.Exec(query, newState, dispatcherID, tenantID, taskID)
|
|
if err != nil {
|
|
return fmt.Errorf("persist task control %q: %w", action, err)
|
|
}
|
|
n, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n != 1 {
|
|
return fmt.Errorf("task control %q rejected for task %q", action, taskID)
|
|
}
|
|
if action == "stop" {
|
|
if _, err := tx.Exec(`UPDATE dispatcher_inbox SET status='suppressed' WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status='pending'`, dispatcherID, tenantID, taskID); err != nil {
|
|
return fmt.Errorf("suppress unstarted stopped-task commands: %w", err)
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit task control %q: %w", action, err)
|
|
}
|
|
return nil
|
|
}
|