package control // Operations stores provider facts independently of a monthly close. Its SQL // transaction is the boundary for observations, projections and collector cursors. import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "encoding/json" "fmt" "net/url" "os" "path/filepath" "time" _ "modernc.org/sqlite" ) type OperationTask struct { Key string `json:"key"` Label string `json:"label"` Owner string `json:"owner"` Phase string `json:"phase"` Scope []string `json:"scope"` Coverage string `json:"coverage"` Gap string `json:"gap"` CanRun bool `json:"can_run"` } type DependencyEdge struct { From string `json:"from"` To string `json:"to"` Kind string `json:"kind"` Evidence string `json:"evidence"` } func OperationCatalog() []OperationTask { return []OperationTask{ {"ar_sync", "AR 源头同步", "Fivetran", "准备输入", []string{"monthly AR connector"}, "partial", "同步历史只读接入；独立产出验证与执行合同未绑定", false}, {"dbt_billing", "财务输入模型", "dbt-dev", "准备输入", []string{"approved billing job"}, "unconfigured", "需绑定明确的 dbt Account / Job", false}, {"crm_extract", "财务 CRM 子集提取", "pingcap-datapipelines", "数据流转", financeCRMTables(true), "partial", "财务前置表观察范围；独立子集执行未绑定，共享进程仅供诊断；dbt额外依赖仍待绑定", false}, {"crm_load", "财务 CRM 逐表加载", "pingcap-datapipelines", "数据流转", financeCRMTables(false), "partial", "逐表审计与当前 manifest 核对；未绑定加载时 generation 或 close 验收；不是全库运行状态", false}, {"salesforce_org", "Org 前置数据", "dataplatform_finance", "出账前置数据", []string{"salesforce_org"}, "partial", "生成/发布审计；执行恢复与目标租约合同待绑定", false}, {"salesforce_account", "Account 前置数据", "dataplatform_finance", "出账前置数据", []string{"salesforce_account"}, "partial", "生成/发布审计；执行恢复与目标租约合同待绑定", false}, {"salesforce_order", "Order 前置数据", "dataplatform_finance", "出账前置数据", []string{"salesforce_order"}, "partial", "需区分既有输入与新 Order 的发布版本", false}, {"run_monthly_cloud_billing", "月度计费", "finance-portal", "计算与拆分", []string{"monthly_revrec_orders", "monthly_revrec_collections"}, "unconfigured", "冻结输入与联合产出验收待接通", false}, {"run_cloud_breakdown", "账单拆分", "finance-portal", "计算与拆分", []string{"monthly_cloud_order", "monthly_cloud_revrec", "monthly_cloud_collection"}, "unconfigured", "精确月份/mode/输入输出版本待接通", false}, {"sync_order_to_crm", "订单同步", "finance-portal", "发布订单", []string{"Order", "OrderItem"}, "unconfigured", "独立验证与部分恢复待接通", false}, {"finalize_revrec", "收入记录", "finance-portal", "完成记录", []string{"monthly_cloud_revrec"}, "unconfigured", "受控业务范围与独立验证待接通", false}, {"finalize_collection", "收款记录", "finance-portal", "完成记录", []string{"monthly_cloud_collection"}, "unconfigured", "受控业务范围与独立验证待接通", false}, } } func OperationDependencies() []DependencyEdge { return []DependencyEdge{ {"ar_sync", "dbt_billing", "data", "v7 target; runtime lineage unbound"}, {"dbt_billing", "run_monthly_cloud_billing", "data", "v7 target; frozen input unbound"}, {"run_monthly_cloud_billing", "run_cloud_breakdown", "data", "v7 target; exact output unbound"}, {"run_cloud_breakdown", "sync_order_to_crm", "control", "v7 target"}, {"sync_order_to_crm", "crm_extract", "data", "v7 target"}, {"crm_extract", "crm_load", "data", "requires identical manifest"}, {"crm_load", "salesforce_order", "data", "v7 target; publication binding unbound"}, {"crm_load", "salesforce_org", "data", "Finance Org SQL reads CRM Org, Account and User; exact input versions unbound"}, {"crm_load", "salesforce_account", "data", "Finance Account SQL reads CRM Account, Org and references; exact input versions unbound"}, {"salesforce_org", "finalize_revrec", "reference", "v7 prerequisite; exact consumer binding pending"}, {"salesforce_account", "finalize_collection", "reference", "v7 prerequisite; exact consumer binding pending"}, {"salesforce_order", "finalize_revrec", "control", "v7 target"}, {"salesforce_order", "finalize_collection", "control", "v7 target"}, } } type ProviderRun struct { ID string `json:"id"` Provider string `json:"provider"` Environment string `json:"environment"` Task string `json:"task"` ExternalID string `json:"external_id"` Status string `json:"status"` RawStatus string `json:"raw_status"` Verification string `json:"verification"` StartedAt time.Time `json:"started_at"` FinishedAt *time.Time `json:"finished_at,omitempty"` EventAt time.Time `json:"event_at"` ObservedAt time.Time `json:"observed_at"` BillMonth string `json:"bill_month,omitempty"` InputRefs []string `json:"input_refs,omitempty"` OutputRefs []string `json:"output_refs,omitempty"` Message string `json:"message"` Manifest *ManifestEvidence `json:"manifest,omitempty"` TimeBasis string `json:"time_basis,omitempty"` } type CollectionState struct { ScopeVersion string `json:"scope_version,omitempty"` Provider string `json:"provider"` Cursor string `json:"cursor"` Status string `json:"status"` Message string `json:"message"` ObservedAt time.Time `json:"observed_at"` CoveredFrom time.Time `json:"covered_from"` } type ScheduleObservation struct { ID string `json:"id"` Provider string `json:"provider"` State string `json:"state"` Expression string `json:"expression"` Timezone string `json:"timezone"` NextAt *time.Time `json:"next_at,omitempty"` ObservedAt time.Time `json:"observed_at"` Coverage string `json:"coverage"` } type TriggerObservation struct { ID string `json:"id"` ScheduleID string `json:"schedule_id"` At time.Time `json:"at"` Status string `json:"status"` ObservedAt time.Time `json:"observed_at"` RunID string `json:"run_id,omitempty"` EvidenceKind string `json:"evidence_kind,omitempty"` Phase string `json:"phase,omitempty"` ScheduledAt *time.Time `json:"scheduled_at,omitempty"` RawStatus string `json:"raw_status,omitempty"` HTTPStatus int `json:"http_status,omitempty"` Message string `json:"message,omitempty"` } type RunBatch struct { Runs []ProviderRun State CollectionState Schedules []ScheduleObservation Triggers []TriggerObservation } type Operations struct { db *sql.DB Environment string } func OpenOperations(path, environment string) (*Operations, error) { if environment == "" { return nil, fmt.Errorf("operations environment is required") } if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { return nil, err } f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0600) if err != nil { return nil, err } f.Close() if err := os.Chmod(path, 0600); err != nil { return nil, err } db, err := sql.Open("sqlite", path) if err != nil { return nil, err } db.SetMaxOpenConns(1) _, err = db.Exec(`PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS metadata(key TEXT PRIMARY KEY,value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY,task TEXT,provider TEXT,status TEXT,started TEXT,event_at TEXT,payload TEXT NOT NULL); CREATE INDEX IF NOT EXISTS runs_order ON runs(started DESC,id DESC); CREATE TABLE IF NOT EXISTS observations(id TEXT PRIMARY KEY,run_id TEXT,event_at TEXT,payload TEXT NOT NULL); CREATE TABLE IF NOT EXISTS checkpoints(provider TEXT PRIMARY KEY,payload TEXT NOT NULL); CREATE TABLE IF NOT EXISTS schedules(id TEXT PRIMARY KEY,payload TEXT NOT NULL); CREATE TABLE IF NOT EXISTS triggers(id TEXT PRIMARY KEY,payload TEXT NOT NULL); CREATE TABLE IF NOT EXISTS alerts(id TEXT PRIMARY KEY,subject TEXT,rule TEXT,status TEXT,opened TEXT,resolved TEXT,actor TEXT DEFAULT '',message TEXT); CREATE UNIQUE INDEX IF NOT EXISTS active_alert ON alerts(subject,rule) WHERE status!='resolved'; CREATE TABLE IF NOT EXISTS alert_events(id INTEGER PRIMARY KEY,alert_id TEXT,at TEXT,action TEXT,actor TEXT); CREATE TABLE IF NOT EXISTS deliveries(id TEXT PRIMARY KEY,alert_id TEXT,status TEXT,attempts INTEGER DEFAULT 0,next_at TEXT,message TEXT);`) if err != nil { db.Close() return nil, err } var existing string err = db.QueryRow("SELECT value FROM metadata WHERE key='environment'").Scan(&existing) if err == sql.ErrNoRows { _, err = db.Exec("INSERT INTO metadata VALUES('environment',?)", environment) } else if err == nil && existing != environment { err = fmt.Errorf("observation store environment mismatch") } if err != nil { db.Close() return nil, err } return &Operations{db: db, Environment: environment}, nil } func (o *Operations) Close() error { return o.db.Close() } func (o *Operations) BindSourceConfig(c LiveSourceConfig) error { financeTarget := "unconfigured" if c.FinanceDatabaseURL != "" { u, err := url.Parse(c.FinanceDatabaseURL) if err != nil { return fmt.Errorf("invalid finance source binding") } financeTarget = u.Host + u.Path } fingerprint := stableID(c.QueryProject, c.CRMControlDataset, c.SchedulerJobName, c.FivetranConnectorID, financeTarget, os.Getenv("FINANCE_PORTAL_DBT_ACCOUNT_ID"), os.Getenv("FINANCE_PORTAL_DBT_JOB_ID"), envOrValue("FINANCE_PORTAL_DBT_API_BASE", "https://cloud.getdbt.com")) legacyFingerprint := fingerprint if c.FinanceDatabaseURL == "" && c.FinanceHistoryAPIURL != "" { fingerprint = stableID(fingerprint, c.FinanceHistoryAPIURL) } var previous string err := o.db.QueryRow("SELECT value FROM metadata WHERE key='source_binding'").Scan(&previous) if err == sql.ErrNoRows { _, err = o.db.Exec("INSERT INTO metadata VALUES('source_binding',?)", fingerprint) return err } if err != nil { return err } if previous != fingerprint { // Add an explicitly configured API only to a previously unbound Finance // namespace. Other provider history and alerts must survive onboarding. if previous == legacyFingerprint && c.FinanceDatabaseURL == "" && c.FinanceHistoryAPIURL != "" { tx, err := o.db.Begin() if err != nil { return err } defer tx.Rollback() var count int if err = tx.QueryRow("SELECT COUNT(*) FROM runs WHERE provider='finance'").Scan(&count); err != nil { return err } if count == 0 { if _, err = tx.Exec("INSERT INTO metadata(key,value) VALUES('finance_api_onboarding_previous_binding',?)", previous); err != nil { return err } if _, err = tx.Exec("UPDATE metadata SET value=? WHERE key='source_binding' AND value=?", fingerprint, previous); err != nil { return err } return tx.Commit() } } return fmt.Errorf("source binding changed; use a separate observation store to preserve history identity") } return nil } func stableID(parts ...string) string { data, _ := json.Marshal(parts) h := sha256.Sum256(data) return hex.EncodeToString(h[:]) } func stamp(t time.Time) string { return t.UTC().Format("2006-01-02T15:04:05.000000000Z07:00") } func terminal(s string) bool { return s == "succeeded" || s == "failed" || s == "cancelled" } func (o *Operations) CommitBatch(ctx context.Context, batch RunBatch) error { if batch.State.Provider == "" { return fmt.Errorf("missing provider namespace") } tx, err := o.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() for _, run := range batch.Runs { if run.ExternalID == "" || run.Task == "" || run.Provider != batch.State.Provider || run.StartedAt.IsZero() || run.EventAt.IsZero() { return fmt.Errorf("incomplete run identity") } run.Environment = o.Environment run.ID = stableID(o.Environment, run.Provider, run.Task, run.ExternalID) run.ObservedAt = batch.State.ObservedAt payload, err := json.Marshal(run) if err != nil { return err } // Poll time is not provider evidence identity. evidence := run evidence.ObservedAt = time.Time{} if run.Manifest != nil { manifest := *run.Manifest manifest.ObservedAt = time.Time{} evidence.Manifest = &manifest } encoded, _ := json.Marshal(evidence) if _, err = tx.ExecContext(ctx, "INSERT OR IGNORE INTO observations VALUES(?,?,?,?)", stableID(string(encoded)), run.ID, stamp(run.EventAt), string(payload)); err != nil { return err } var previous string err = tx.QueryRowContext(ctx, "SELECT payload FROM runs WHERE id=?", run.ID).Scan(&previous) if err != nil && err != sql.ErrNoRows { return err } if previous != "" { var old ProviderRun if err = json.Unmarshal([]byte(previous), &old); err != nil { return err } if run.EventAt.Before(old.EventAt) || (terminal(old.Status) && !terminal(run.Status)) { continue } } _, err = tx.ExecContext(ctx, `INSERT INTO runs VALUES(?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET status=excluded.status,event_at=excluded.event_at,payload=excluded.payload`, run.ID, run.Task, run.Provider, run.Status, stamp(run.StartedAt), stamp(run.EventAt), string(payload)) if err != nil { return err } if run.Status == "failed" { if err = openAlert(ctx, tx, run.ID, "execution_failed", run.Task+"：任务执行失败", batch.State.ObservedAt); err != nil { return err } } // Only a correction to the same provider run can resolve that run's failure. if run.Status == "succeeded" { if err = resolveAlert(ctx, tx, run.ID, "execution_failed", batch.State.ObservedAt); err != nil { return err } } } for _, s := range batch.Schedules { p, _ := json.Marshal(s) if _, err = tx.ExecContext(ctx, "INSERT INTO schedules VALUES(?,?) ON CONFLICT(id) DO UPDATE SET payload=excluded.payload", s.ID, string(p)); err != nil { return err } } for _, t := range batch.Triggers { p, _ := json.Marshal(t) if _, err = tx.ExecContext(ctx, "INSERT INTO triggers VALUES(?,?) ON CONFLICT(id) DO UPDATE SET payload=excluded.payload", t.ID, string(p)); err != nil { return err } if t.Status == "failed" { rule, message := "submission_failed", "调度请求失败；需结合任务审计区分提交与执行失败" if t.EvidenceKind == "scheduler_log" { rule, message = "scheduler_response_failed", "Scheduler 响应日志记录失败；实际任务结果与运行关联需独立核验" } if err = openAlert(ctx, tx, t.ID, rule, message, t.ObservedAt); err != nil { return err } } } p, _ := json.Marshal(batch.State) if _, err = tx.ExecContext(ctx, "INSERT INTO checkpoints VALUES(?,?) ON CONFLICT(provider) DO UPDATE SET payload=excluded.payload", batch.State.Provider, string(p)); err != nil { return err } if batch.State.Status == "unavailable" { err = openAlert(ctx, tx, batch.State.Provider, "collection_unavailable", batch.State.Message, batch.State.ObservedAt) } else if batch.State.Status == "partial" || batch.State.Status == "healthy" { err = resolveAlert(ctx, tx, batch.State.Provider, "collection_unavailable", batch.State.ObservedAt) } if err != nil { return err } return tx.Commit() } func openAlert(ctx context.Context, tx *sql.Tx, subject, rule, message string, at time.Time) error { id := stableID(subject, rule, stamp(at)) r, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO alerts(id,subject,rule,status,opened,message) VALUES(?,?,?,'open',?,?)", id, subject, rule, stamp(at), message) if err != nil { return err } n, _ := r.RowsAffected() if n == 0 { return nil } if _, err = tx.ExecContext(ctx, "INSERT INTO alert_events(alert_id,at,action,actor) VALUES(?,?,'opened','collector')", id, stamp(at)); err != nil { return err } _, err = tx.ExecContext(ctx, "INSERT INTO deliveries(id,alert_id,status,next_at,message) VALUES(?,?,'pending',?,'通知渠道未配置')", id, id, stamp(at)) return err } func resolveAlert(ctx context.Context, tx *sql.Tx, subject, rule string, at time.Time) error { var id string err := tx.QueryRowContext(ctx, "SELECT id FROM alerts WHERE subject=? AND rule=? AND status!='resolved'", subject, rule).Scan(&id) if err == sql.ErrNoRows { return nil } if err != nil { return err } if _, err = tx.ExecContext(ctx, "UPDATE alerts SET status='resolved',resolved=? WHERE id=?", stamp(at), id); err != nil { return err } _, err = tx.ExecContext(ctx, "INSERT INTO alert_events(alert_id,at,action,actor) VALUES(?,?,'resolved','collector')", id, stamp(at)) return err } func (o *Operations) Checkpoint(ctx context.Context, provider string) CollectionState { var payload string var state CollectionState if o.db.QueryRowContext(ctx, "SELECT payload FROM checkpoints WHERE provider=?", provider).Scan(&payload) == nil { _ = json.Unmarshal([]byte(payload), &state) } return state } type RunFilter struct { Task, Provider, Status, BeforeID, BeforeTime string OutputRef string Limit int } func (o *Operations) Runs(ctx context.Context, f RunFilter) ([]ProviderRun, error) { if f.Limit < 1 || f.Limit > 100 { f.Limit = 50 } rows, err := o.db.QueryContext(ctx, `SELECT payload FROM runs WHERE (?='' OR task=?) AND (?='' OR provider=?) AND (?='' OR status=?) AND (?='' OR EXISTS (SELECT 1 FROM json_each(runs.payload,'$.output_refs') WHERE value=?)) AND (?='' OR started