package domain import ( "crypto/rand" "crypto/sha256" "encoding/base32" "encoding/binary" "encoding/hex" "encoding/json" "errors" "fmt" "strings" "time" ) var ErrConflict = errors.New("task version conflict") var ErrNotFound = errors.New("task not found") var ErrInvalid = errors.New("invalid event") // ErrDuplicate is returned by Store.Append for a TaskCreated event whose // (source, external_id) pair was already ingested. The caller already has a // task for this content; nothing was appended. var ErrDuplicate = errors.New("duplicate task ingestion") // CurrentEventSchema is 2: schema 2 requires every event to declare its // authorizing Surface (see ValidateEvent), enforced at the store append // boundary. Schema 1 events already on disk replay unchanged — tolerant // reader, not upcast (spec open question #2). const CurrentEventSchema = 2 type TaskState string const ( StateQueued TaskState = "queued" StateLeased TaskState = "leased" StateCompleted TaskState = "completed" StateFailed TaskState = "failed" StateBlocked TaskState = "blocked" ) type Estimate struct { Value float64 `json:"value"` Who string `json:"who"` Confidence float64 `json:"confidence"` } type Lease struct { HarnessID string `json:"harness_id"` Until time.Time `json:"until"` } type Task struct { ID string `json:"id"` Source string `json:"source"` ExternalID string `json:"external_id"` Project string `json:"project"` Capability []string `json:"capability"` Parent string `json:"parent,omitempty"` InherentPriority int `json:"inherent_priority"` Due *time.Time `json:"due,omitempty"` Estimate *Estimate `json:"estimate,omitempty"` State TaskState `json:"state"` Lease *Lease `json:"lease,omitempty"` Version int `json:"version"` Title string `json:"title,omitempty"` Description string `json:"description,omitempty"` } type Event struct { SchemaVersion int `json:"schema_version,omitempty"` Seq uint64 `json:"seq"` ID string `json:"id"` Type string `json:"type"` TaskID string `json:"task_id"` Version int `json:"version"` At time.Time `json:"at"` Payload json.RawMessage `json:"payload"` // Surface identifies the bus capability the emitter is authorized under // (see internal/authz). It is required on every event so authorization is // enforced once, at the store append boundary, regardless of whether the // emitter reached the store over HTTP, from the router, from a harness // adapter, or from a provider. Surface string `json:"surface"` } func Hash(v []byte) string { h := sha256.Sum256(v); return hex.EncodeToString(h[:]) } // NewID returns a sortable, 128-bit ULID-like identifier using the canonical // 48-bit millisecond timestamp plus 80 bits of cryptographic randomness. var ulidEncoding = base32.NewEncoding("0123456789ABCDEFGHJKMNPQRSTVWXYZ").WithPadding(base32.NoPadding) func NewID() string { b := make([]byte, 16) binary.BigEndian.PutUint64(b[:8], uint64(time.Now().UnixMilli())<<16) _, _ = rand.Read(b[6:]) return ulidEncoding.EncodeToString(b) } func ValidateEvent(e Event) error { if e.SchemaVersion > CurrentEventSchema || e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 { return ErrInvalid } if e.SchemaVersion >= 2 && strings.TrimSpace(e.Surface) == "" { return fmt.Errorf("%w: surface required", ErrInvalid) } allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true} if !allowed[e.Type] { return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type) } var p map[string]any if err := json.Unmarshal(e.Payload, &p); err != nil { return fmt.Errorf("%w: payload is not JSON", ErrInvalid) } if p == nil { return fmt.Errorf("%w: payload must be an object", ErrInvalid) } return ValidatePayload(e.Type, p) } func ValidateCreated(p map[string]any) error { for _, k := range []string{"source", "external_id", "project"} { if s, ok := p[k].(string); !ok || strings.TrimSpace(s) == "" { return fmt.Errorf("%w: %s required", ErrInvalid, k) } } return nil } func ValidatePayload(typ string, p map[string]any) error { requiredString := func(key string) error { v, ok := p[key].(string) if !ok || strings.TrimSpace(v) == "" { return fmt.Errorf("%w: %s required", ErrInvalid, key) } return nil } switch typ { case "TaskCreated": return ValidateCreated(p) case "TaskLeased": if err := requiredString("harness_id"); err != nil { return err } until, untilOK := p["until_ns"].(float64) if ttl, ok := p["ttl"].(float64); ok { if ttl <= 0 { return fmt.Errorf("%w: ttl invalid", ErrInvalid) } } else if !untilOK || until <= float64(time.Now().UnixNano()) { return fmt.Errorf("%w: ttl required", ErrInvalid) } if v, ok := p["expected_version"].(float64); !ok || v < 0 || v != float64(int(v)) { return fmt.Errorf("%w: expected_version invalid", ErrInvalid) } case "TaskReleased": if err := requiredString("handoff_ref"); err != nil && p["reason"] == nil { return err } if _, ok := p["handoff_ref"]; ok { if err := requiredHash(p, "handoff_ref"); err != nil { return err } v, ok := p["anchor_sha"].(string) if !ok || len(v) != 40 || strings.TrimSpace(v) != v { return fmt.Errorf("%w: anchor_sha invalid", ErrInvalid) } } case "TaskCompleted": if err := requiredString("report_ref"); err != nil { return err } if err := requiredHash(p, "report_ref"); err != nil { return err } if receipt, ok := p["receipt"].(map[string]any); !ok || len(receipt) == 0 { return fmt.Errorf("%w: receipt required", ErrInvalid) } case "TaskFailed": if err := requiredString("reason"); err != nil { return err } case "TaskBlocked": if err := requiredString("blocker"); err != nil { return err } if _, ok := p["handoff_ref"]; ok { if err := requiredHash(p, "handoff_ref"); err != nil { return err } } case "TaskAmended": if len(p) == 0 { return fmt.Errorf("%w: amendment cannot be empty", ErrInvalid) } case "TaskCorrected": // §3.1: "a wrong event is never edited; a compensating event is // appended and replay sees both." `corrects` names the event this one // reverses/repairs — existence against the log is checked in // Store.Append, where the log is visible; ValidatePayload only knows // shape. if err := requiredString("corrects"); err != nil { return err } if v, ok := p["state"]; ok { s, ok := v.(string) if !ok { return fmt.Errorf("%w: state must be a string", ErrInvalid) } switch TaskState(s) { case StateQueued, StateLeased, StateCompleted, StateFailed, StateBlocked: default: return fmt.Errorf("%w: state invalid", ErrInvalid) } } if len(p) < 2 { return fmt.Errorf("%w: correction must change at least one field", ErrInvalid) } case "ApprovalRequested": for _, k := range []string{"subject_ref", "options"} { if _, ok := p[k]; !ok { return fmt.Errorf("%w: %s required", ErrInvalid, k) } } case "ApprovalGranted", "ApprovalDenied": if err := requiredString("subject_ref"); err != nil { return err } case "QuotaReported": if err := requiredString("harness_id"); err != nil { return err } if v, ok := p["consumed"].(float64); !ok || v < 0 { return fmt.Errorf("%w: consumed required", ErrInvalid) } case "StandupAdvisory": if _, ok := p["items"]; !ok { return fmt.Errorf("%w: items required", ErrInvalid) } } return nil } func requiredHash(p map[string]any, key string) error { v, ok := p[key].(string) if !ok || len(v) != 64 { return fmt.Errorf("%w: %s must be sha256", ErrInvalid, key) } if _, err := hex.DecodeString(v); err != nil { return fmt.Errorf("%w: %s must be sha256", ErrInvalid, key) } return nil }