diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index a17b6dc..457a598 100644 --- a/internal/cardsvc/cardsvc.go +++ b/internal/cardsvc/cardsvc.go @@ -27,8 +27,9 @@ var ( const cardAppendWindow = 250 * time.Millisecond type Service struct { - st *store.Store - emit Emitter + TaskCommitter func(store.TaskProjectionWrite) ([]core.StreamEvent, error) + st *store.Store + emit Emitter // MirrorConvCards EXT-12:会话卡是否镜像进项目公共流(缺省 true=现状全量 // 镜像;false 时会话卡只进本人会话流,项目级卡不受影响)。 MirrorConvCards bool @@ -97,6 +98,13 @@ func (s *Service) List(owner, conversationID string) ([]core.ActivityCard, error // Update 保存整卡变更(流式 Body 增量等)并发布 card.updated;ref 保持原值。 // 系统镜像路径(同会话 run 的顺序行为)沿用盲写;用户态裁决走 UpdateVersioned。 func (s *Service) Update(owner string, card core.ActivityCard) error { + if card.Task != nil { + _, ver, err := s.st.GetCardVersion(owner, card.ID) + if err != nil { + return err + } + return s.UpdateVersioned(owner, card, ver) + } card.UpdatedAt = s.now().UTC() if err := s.st.SaveCard(owner, &card, ""); err != nil { return err @@ -107,6 +115,9 @@ func (s *Service) Update(owner string, card core.ActivityCard) error { // UpdateVersioned 乐观并发保存(U3,KTD7):版本不匹配透传 store.ErrConflict, // HTTP 层映射 409;客户端重读卡后重试。 func (s *Service) UpdateVersioned(owner string, card core.ActivityCard, expectedVersion int) error { + if card.Task != nil { + return s.commitProjection(store.TaskProjectionWrite{Owner: owner, Card: card, ExpectedCardVersion: expectedVersion, ExpectedTaskRevision: card.Task.Revision, Channels: s.taskChannels(card)}) + } card.UpdatedAt = s.now().UTC() if err := s.st.SaveCardVersioned(owner, &card, "", expectedVersion); err != nil { return err @@ -123,6 +134,9 @@ func (s *Service) Append(owner, id, text string) error { if err != nil { return err } + if c.Task != nil { + return ErrNoAction + } if text != "" { if n := len(c.Body); n > 0 && c.Body[n-1].Type == core.UIMarkdown { c.Body[n-1].Markdown += text @@ -227,6 +241,9 @@ func (s *Service) ToProject(owner, id string) (core.ActivityCard, error) { if c.ConversationID == "" { return c, errors.New("card: already project-level") } + if c.Task != nil { + return c, errors.New("card: keep the original task reference in its conversation") + } if ref, rerr := s.st.CardRef(owner, c.ID); rerr == nil { if strings.HasPrefix(ref, "hitl:") || strings.HasPrefix(ref, "cmd:") { return c, errors.New("card: flow-bound card cannot be archived to project") diff --git a/internal/cardsvc/task_projection.go b/internal/cardsvc/task_projection.go new file mode 100644 index 0000000..cd0ffd8 --- /dev/null +++ b/internal/cardsvc/task_projection.go @@ -0,0 +1,59 @@ +package cardsvc + +import ( + "agent-system-backend/internal/store" + core "agent-system-core" + "time" +) + +func taskCardState(task core.TaskSnapshot) (core.CardStatus, string) { + switch task.ExecutionState { + case "completed": + return core.CardDone, "ok" + case "failed", "interrupted": + return core.CardFailed, "danger" + case "awaiting_user": + return core.CardActionRequired, "warn" + default: + return core.CardActive, "info" + } +} +func (s *Service) commitProjection(w store.TaskProjectionWrite) error { + if s.TaskCommitter != nil { + _, err := s.TaskCommitter(w) + return err + } + _, err := s.st.CommitTaskProjection(w) + return err +} +func (s *Service) taskChannels(card core.ActivityCard) []string { + channels := []string{card.ConversationID} + if s.MirrorConvCards && card.ProjectID != "" { + channels = append(channels, string(core.ProjectChannelKey(card.ProjectID))) + } + return channels +} +func (s *Service) CommitTask(owner string, next core.TaskSnapshot, expectedRevision int64) error { + c, version, err := s.st.GetCardVersion(owner, next.CardID) + if err != nil { + return err + } + c.Task = &next + c.Body = next.SummaryBody() + c.Status, c.Tone = taskCardState(next) + c.UpdatedAt = time.Now().UTC() + return s.commitProjection(store.TaskProjectionWrite{Owner: owner, Card: c, ExpectedCardVersion: version, ExpectedTaskRevision: expectedRevision, Channels: s.taskChannels(c)}) +} +func (s *Service) CreateTask(owner string, card core.ActivityCard, ref string) (core.ActivityCard, error) { + if card.Task == nil { + return card, store.ErrConflict + } + card.Status, card.Tone = taskCardState(*card.Task) + card.Body = card.Task.SummaryBody() + card.CreatedAt = time.Now().UTC() + card.UpdatedAt = card.CreatedAt + if err := s.commitProjection(store.TaskProjectionWrite{Owner: owner, Card: card, Ref: ref, Channels: s.taskChannels(card)}); err != nil { + return card, err + } + return s.st.GetCard(owner, card.ID) +} diff --git a/internal/cardsvc/task_projection_test.go b/internal/cardsvc/task_projection_test.go new file mode 100644 index 0000000..96b7419 --- /dev/null +++ b/internal/cardsvc/task_projection_test.go @@ -0,0 +1,63 @@ +package cardsvc + +import ( + core "agent-system-core" + "testing" +) + +func TestTaskProjectionPreservesUserReadingPreferences(t *testing.T) { + st, svc, _ := openTest(t) + c := core.ActivityCard{ID: "card-task", ProjectID: "p1", ConversationID: "c1", Title: "任务", Kind: core.CardKindProgress} + c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: "p1", ConversationID: "c1", OwnerInstanceID: "i1", StreamScope: "conversation:c1", Revision: 1, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "任务"} + if _, e := svc.CreateTask("o", c, ""); e != nil { + t.Fatal(e) + } + current, e := st.GetCard("o", c.ID) + if e != nil { + t.Fatal("task card not saved", e) + } + current.Pinned = true + current.Status = core.CardDismissed + if e = st.SaveCard("o", ¤t, ""); e != nil { + t.Fatal(e) + } + next := *current.Task + next.Revision = 2 + next.ExecutionState = "completed" + next.Conclusion = "有用成果" + if e = svc.CommitTask("o", next, 1); e != nil { + t.Fatal(e) + } + got, e := st.GetCard("o", c.ID) + if e != nil || !got.Pinned || got.Status != core.CardDismissed || got.Task.ExecutionState != "completed" { + t.Fatal("work update stole user state", got, e) + } +} + +func TestTaskProjectionUserPinHasAtomicEventWatermark(t *testing.T) { + st, svc, _ := openTest(t) + c := core.ActivityCard{ID: "card-pin", ProjectID: "p1", ConversationID: "c1", Title: "任务", Kind: core.CardKindProgress} + c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: "p1", ConversationID: "c1", OwnerInstanceID: "i1", StreamScope: "conversation:c1", Revision: 1, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "任务"} + if _, e := svc.CreateTask("o", c, ""); e != nil { + t.Fatal(e) + } + if _, e := svc.SetPinned("o", c.ID, true); e != nil { + t.Fatal(e) + } + snapshot, e := st.CardsAtCursor("o", "c1") + if e != nil || snapshot.ReplaySeq != 2 || len(snapshot.Cards) != 1 || !snapshot.Cards[0].Pinned { + t.Fatal("pin not atomically replayable", snapshot, e) + } +} + +func TestTaskProjectionCannotCreateSecondExecutionOwner(t *testing.T) { + _, svc, _ := openTest(t) + c := core.ActivityCard{ID: "card-archive", ProjectID: "p1", ConversationID: "c1", Title: "任务", Kind: core.CardKindProgress} + c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: "p1", ConversationID: "c1", OwnerInstanceID: "i1", StreamScope: "conversation:c1", Revision: 1, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "任务"} + if _, e := svc.CreateTask("o", c, ""); e != nil { + t.Fatal(e) + } + if _, e := svc.ToProject("o", c.ID); e == nil { + t.Fatal("task was copied with a second execution owner") + } +} diff --git a/internal/runmgr/events.go b/internal/runmgr/events.go index 9e41162..5511c9f 100644 --- a/internal/runmgr/events.go +++ b/internal/runmgr/events.go @@ -157,3 +157,40 @@ func (eb *eventBuffer) flushAll() { eb.flushConversation(id) } } + +// commitTask keeps buffered records and the work projection in one transaction. +func (eb *eventBuffer) commitTask(w store.TaskProjectionWrite) ([]core.StreamEvent, error) { + eb.flushing.Lock() + defer eb.flushing.Unlock() + channels := append([]string(nil), w.Channels...) + if len(channels) == 0 { + channels = []string{w.Card.ConversationID} + } + eb.mu.Lock() + for _, ch := range channels { + w.Buffered = append(w.Buffered, eb.buf[ch]...) + delete(eb.buf, ch) + } + eb.mu.Unlock() + events, err := eb.store.CommitTaskProjection(w) + if err != nil { + eb.mu.Lock() + for i := len(w.Buffered) - 1; i >= 0; i-- { + e := w.Buffered[i] + eb.buf[e.ConversationID] = append([]core.StreamEvent{e}, eb.buf[e.ConversationID]...) + } + eb.mu.Unlock() + return nil, err + } + grouped := map[string][]core.StreamEvent{} + for _, e := range events { + eb.bus.Publish(e) + grouped[e.ConversationID] = append(grouped[e.ConversationID], e) + } + if eb.onPersisted != nil { + for ch, batch := range grouped { + eb.onPersisted(ch, batch) + } + } + return events, nil +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 1086b73..0ee65a8 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -180,7 +180,12 @@ type IntentGate interface { func (m *Manager) SetGate(g IntentGate) { m.gate = g } // SetCards 注入收件箱卡服务(nil = 不产卡;T0 退回基线仍成立)。 -func (m *Manager) SetCards(s *cardsvc.Service) { m.cards = s } +func (m *Manager) SetCards(s *cardsvc.Service) { + m.cards = s + if s != nil { + s.TaskCommitter = m.events.commitTask + } +} // EmitCardEvent 供服务层(cardsvc 投影)复用会话事件管道(持久化 + seq + 扇出)。 func (m *Manager) EmitCardEvent(conversationID, runID string, typ core.StreamEventType, payload any) error { diff --git a/internal/server/server.go b/internal/server/server.go index 2cdacd7..5c15c99 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1144,8 +1144,8 @@ func (s *Server) listCards(w http.ResponseWriter, r *http.Request) { writeStoreErr(w, err) return } - cards, err := s.Cards.List(owner, r.PathValue("id")) - writeJSONOr(w, map[string]any{"cards": cards}, err) + snapshot, err := s.st.CardsAtCursor(owner, r.PathValue("id")) + writeJSONOr(w, snapshot, err) } // listProjectCards 项目收件箱聚合(D49 拍板口径):项目级卡 ∪ 该项目全部会话 diff --git a/internal/store/messages.go b/internal/store/messages.go index 36c2042..4216b9f 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -274,6 +274,21 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int return 0, err } defer tx.Rollback() + base, err := s.appendEventsTx(tx, conversationID, evs) + if err != nil { + return 0, err + } + if err = tx.Commit(); err != nil { + return 0, err + } + return base, nil +} + +// appendEventsTx shares sequence allocation and transactional NOTIFY with projections. +func (s *Store) appendEventsTx(tx *sql.Tx, conversationID string, evs []core.StreamEvent) (int64, error) { + if len(evs) == 0 { + return 0, nil + } if err := s.dialect.lockConversation(tx, conversationID); err != nil { return 0, err } @@ -306,9 +321,6 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int return 0, err } } - if err := tx.Commit(); err != nil { - return 0, err - } for i := range evs { evs[i].Seq = base.Int64 + int64(i) + 1 } diff --git a/internal/store/task_projection.go b/internal/store/task_projection.go new file mode 100644 index 0000000..aba0a00 --- /dev/null +++ b/internal/store/task_projection.go @@ -0,0 +1,226 @@ +package store + +import ( + core "agent-system-core" + "bytes" + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "sort" + "time" +) + +type TaskProjectionWrite struct { + Owner string + Card core.ActivityCard + Ref string + ExpectedCardVersion int + ExpectedTaskRevision int64 + Channels []string + Buffered []core.StreamEvent +} +type CardsSnapshot struct { + Cards []core.ActivityCard `json:"cards"` + ReplaySeq int64 `json:"replaySeq"` +} + +func (s *Store) CommitTaskProjection(w TaskProjectionWrite) ([]core.StreamEvent, error) { + c := w.Card + if c.Task == nil || c.Task.CardID != c.ID || c.Task.ConversationID != c.ConversationID || c.Task.ProjectID != c.ProjectID { + return nil, fmt.Errorf("task.scope: card binding mismatch") + } + task := *c.Task + c.Task = &task + if err := task.Validate(); err != nil { + return nil, err + } + channels := append([]string(nil), w.Channels...) + if len(channels) == 0 { + channels = []string{c.ConversationID} + } + sort.Strings(channels) + unique := channels[:0] + for _, ch := range channels { + if ch != c.ConversationID && ch != "project:"+c.ProjectID { + return nil, fmt.Errorf("task.scope: foreign channel") + } + if len(unique) == 0 || unique[len(unique)-1] != ch { + unique = append(unique, ch) + } + } + channels = unique + tx, err := s.db.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + for _, ch := range channels { + if err = s.dialect.lockConversation(tx, ch); err != nil { + return nil, err + } + } + var oldPayload, oldOwner string + var version int + err = tx.QueryRow(s.q(`SELECT payload,owner,version FROM activity_cards WHERE id=?`), c.ID).Scan(&oldPayload, &oldOwner, &version) + creating := errors.Is(err, sql.ErrNoRows) + viewOnly := false + if err != nil && !creating { + return nil, err + } + if creating { + if w.ExpectedCardVersion != 0 || w.ExpectedTaskRevision != 0 || task.Revision != 1 { + return nil, ErrConflict + } + } else { + if oldOwner != w.Owner { + return nil, ErrNotFound + } + var old core.ActivityCard + if err = json.Unmarshal([]byte(oldPayload), &old); err != nil { + return nil, err + } + if version != w.ExpectedCardVersion || old.Task == nil || old.Task.Revision != w.ExpectedTaskRevision || (task.Revision != old.Task.Revision+1 && task.Revision != old.Task.Revision) || old.Task.TaskID != task.TaskID || old.Task.AttemptID != task.AttemptID || old.ConversationID != c.ConversationID || old.ProjectID != c.ProjectID || old.Task.OwnerInstanceID != task.OwnerInstanceID { + return nil, ErrConflict + } + if task.Revision == old.Task.Revision { + a, _ := json.Marshal(old.Task) + b, _ := json.Marshal(task) + if !bytes.Equal(a, b) { + return nil, ErrConflict + } + viewOnly = true + } + if old.Task.Terminal() && old.Task.ExecutionState != task.ExecutionState { + return nil, ErrConflict + } + if old.Task.RunID != "" && old.Task.RunID != task.RunID { + return nil, ErrConflict + } + if task.ExecutionEpoch < old.Task.ExecutionEpoch { + return nil, ErrConflict + } + if !viewOnly { + c.Pinned = old.Pinned + if old.Status == core.CardDismissed || old.Status == core.CardResolved { + c.Status = old.Status + } + } + } + var events []core.StreamEvent + for _, ch := range channels { + var batch []core.StreamEvent + for _, e := range w.Buffered { + if e.ConversationID == ch { + batch = append(batch, e) + } + } + if len(batch) > 0 { + if _, err = s.appendEventsTx(tx, ch, batch); err != nil { + return nil, err + } + events = append(events, batch...) + } + } + if !viewOnly { + if err = tx.QueryRow(s.q(`SELECT COALESCE(MAX(seq),0)+1 FROM events WHERE conversation_id=?`), c.ConversationID).Scan(&task.SourceSeq); err != nil { + return nil, err + } + } + raw, err := json.Marshal(c) + if err != nil { + return nil, err + } + now := time.Now().UTC() + c.UpdatedAt = now + if c.CreatedAt.IsZero() { + c.CreatedAt = now + } + raw, err = json.Marshal(c) + if err != nil { + return nil, err + } + if creating { + _, err = tx.Exec(s.q(`INSERT INTO activity_cards(id,project_id,conversation_id,ref,owner,payload,status,pinned,created_at,updated_at,version) VALUES(?,?,?,?,?,?,?,?,?,?,1)`), c.ID, c.ProjectID, c.ConversationID, w.Ref, w.Owner, string(raw), c.Status, boolInt(c.Pinned), c.CreatedAt.Format(time.RFC3339Nano), c.UpdatedAt.Format(time.RFC3339Nano)) + } else { + var res sql.Result + res, err = tx.Exec(s.q(`UPDATE activity_cards SET payload=?,status=?,pinned=?,updated_at=?,version=version+1 WHERE id=? AND owner=? AND version=?`), string(raw), c.Status, boolInt(c.Pinned), c.UpdatedAt.Format(time.RFC3339Nano), c.ID, w.Owner, version) + if err == nil { + n, _ := res.RowsAffected() + if n != 1 { + return nil, ErrConflict + } + } + } + if err != nil { + if isUniqueViolation(err) { + return nil, ErrConflict + } + return nil, err + } + payload, err := json.Marshal(core.CardPayload{Card: c}) + if err != nil { + return nil, err + } + for _, ch := range channels { + e := core.StreamEvent{ConversationID: ch, RunID: task.RunID, TS: now.Format(time.RFC3339Nano), V: core.ProtocolVersion, Type: core.EvCardUpdated, Payload: payload} + if creating { + e.Type = core.EvCardCreated + } + if pid, ok := core.ProjectFromChannel(core.ID(ch)); ok { + e.ProjectID = pid + } + batch := []core.StreamEvent{e} + if _, err = s.appendEventsTx(tx, ch, batch); err != nil { + return nil, err + } + events = append(events, batch...) + } + if err = tx.Commit(); err != nil { + return nil, err + } + return events, nil +} +func (s *Store) CardsAtCursor(owner, conversationID string) (CardsSnapshot, error) { + opts := &sql.TxOptions{} + if s.Dialect() == "postgres" { + opts.Isolation = sql.LevelRepeatableRead + opts.ReadOnly = true + } + tx, err := s.db.BeginTx(context.Background(), opts) + if err != nil { + return CardsSnapshot{}, err + } + defer tx.Rollback() + rows, err := tx.Query(s.q(`SELECT payload FROM activity_cards WHERE conversation_id=? AND owner=? ORDER BY updated_at DESC`), conversationID, owner) + if err != nil { + return CardsSnapshot{}, err + } + out := CardsSnapshot{Cards: []core.ActivityCard{}} + for rows.Next() { + var raw string + if err = rows.Scan(&raw); err != nil { + rows.Close() + return out, err + } + var c core.ActivityCard + if err = json.Unmarshal([]byte(raw), &c); err != nil { + rows.Close() + return out, err + } + out.Cards = append(out.Cards, c) + } + err = rows.Err() + rows.Close() + if err != nil { + return out, err + } + if err = tx.QueryRow(s.q(`SELECT COALESCE(MAX(seq),0) FROM events WHERE conversation_id=?`), conversationID).Scan(&out.ReplaySeq); err != nil { + return out, err + } + if err = tx.Commit(); err != nil { + return out, err + } + return out, nil +} diff --git a/internal/store/task_projection_test.go b/internal/store/task_projection_test.go new file mode 100644 index 0000000..69024c2 --- /dev/null +++ b/internal/store/task_projection_test.go @@ -0,0 +1,92 @@ +package store + +import ( + core "agent-system-core" + "errors" + "testing" + "time" +) + +func projectionFixture(t *testing.T, st *Store) core.ActivityCard { + t.Helper() + p, e := st.CreateProject("o", "projection") + if e != nil { + t.Fatal(e) + } + conv, e := st.CreateConversation("o", p.ID, "projection", "") + if e != nil { + t.Fatal(e) + } + c := core.ActivityCard{ID: NewID("card"), ProjectID: p.ID, ConversationID: conv.ID, Kind: core.CardKindProgress, Title: "整理", Status: core.CardActive, CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC()} + c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: p.ID, ConversationID: conv.ID, OwnerInstanceID: "i1", Revision: 1, StreamScope: "conversation:" + conv.ID, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "整理"} + c.Body = c.Task.SummaryBody() + return c +} +func TestTaskProjectionCommitsCardAndBothStreams(t *testing.T) { + st := openTestStore(t) + c := projectionFixture(t, st) + events, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c, Channels: []string{c.ConversationID, "project:" + c.ProjectID}}) + if e != nil || len(events) != 2 { + t.Fatal("missing atomic projection", events, e) + } + got, e := st.GetCard("o", c.ID) + if e != nil || got.Task == nil || got.Task.SourceSeq != 1 { + t.Fatal("card watermark not committed", e) + } + for _, channel := range []string{c.ConversationID, "project:" + c.ProjectID} { + rows, e := st.EventsAfter(channel, 0, 20) + if e != nil || len(rows) != 1 { + t.Fatal("event missing", e) + } + } + snapshot, e := st.CardsAtCursor("o", c.ConversationID) + if e != nil || snapshot.ReplaySeq != 1 || len(snapshot.Cards) != 1 { + t.Fatal("inconsistent snapshot", snapshot, e) + } +} +func TestTaskProjectionRejectsOwnerOldRevisionAndLateState(t *testing.T) { + st := openTestStore(t) + c := projectionFixture(t, st) + if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c}); e != nil { + t.Fatal(e) + } + next := c + copy := *c.Task + next.Task = © + copy.Revision = 2 + copy.ExecutionState = "completed" + next.Status = core.CardDone + if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "other", Card: next, ExpectedCardVersion: 1, ExpectedTaskRevision: 1}); e == nil { + t.Fatal("foreign write accepted") + } + if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: next, ExpectedCardVersion: 1, ExpectedTaskRevision: 1}); e != nil { + t.Fatal(e) + } + copy.Revision = 3 + copy.ExecutionState = "running" + copy.RunID = "run1" + if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: next, ExpectedCardVersion: 2, ExpectedTaskRevision: 2}); !errors.Is(e, ErrConflict) { + t.Fatal("terminal regressed", e) + } +} +func TestTaskProjectionEventFailureRollsBackCard(t *testing.T) { + st := openTestStore(t) + c := projectionFixture(t, st) + var injection = `CREATE TRIGGER reject_task_event BEFORE INSERT ON events BEGIN SELECT RAISE(ABORT,'event injection'); END` + if st.Dialect() == "postgres" { + if _, e := st.db.Exec(`CREATE FUNCTION reject_task_event_fn() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'event injection'; END $$`); e != nil { + t.Fatal(e) + } + injection = `CREATE TRIGGER reject_task_event BEFORE INSERT ON events FOR EACH ROW EXECUTE FUNCTION reject_task_event_fn()` + } + if _, e := st.db.Exec(injection); e != nil { + t.Fatal(e) + } + _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c}) + if e == nil { + t.Fatal("injection did not fail") + } + if _, e = st.GetCard("o", c.ID); !errors.Is(e, ErrNotFound) { + t.Fatal("card survived rolled-back event", e) + } +}