From 07b4e63366ff2c3859262992a56d59de4fa46347 Mon Sep 17 00:00:00 2001 From: toki Date: Sun, 14 Jun 2026 06:04:03 +0900 Subject: [PATCH] feat: operation event outbox milestone implementation - Add operation event store schema and migrations - Implement event persistence in storage layer - Update event domain models for outbox pattern - Add contract notes for control plane event flow - Complete milestone documentation --- .../milestones/operation-event-outbox.md | 8 +- .../code_review_cloud_G05_0.log | 203 ++++++++++++ .../code_review_cloud_G05_1.log | 194 ++++++++++++ .../03+01_operation_event_store/complete.log | 44 +++ .../plan_local_G06_0.log | 295 ++++++++++++++++++ .../plan_local_G06_1.log | 224 +++++++++++++ .../contracts/notes/gito-control-plane.md | 41 +++ services/core/internal/events/events.go | 9 +- services/core/internal/storage/postgres.go | 170 +++++++++- .../core/internal/storage/postgres_test.go | 262 ++++++++++++++++ services/core/internal/storage/storage.go | 3 + .../core/internal/storage/storage_test.go | 9 + services/core/migrations/00001_initial.sql | 10 +- 13 files changed, 1463 insertions(+), 9 deletions(-) create mode 100644 agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_0.log create mode 100644 agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_1.log create mode 100644 agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/complete.log create mode 100644 agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_0.log create mode 100644 agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_1.log diff --git a/agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md b/agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md index efbb9f5..bbcfa8f 100644 --- a/agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md +++ b/agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md @@ -37,16 +37,16 @@ operation lifecycle과 normalized event를 PostgreSQL 원장에 기록하고, pr Git operation, agent run, provider projection을 같은 operation 원장으로 관리한다. - [x] [op-model] operation type/state/idempotency schema를 확정한다. -- [ ] [op-store] operation create/start/succeed/fail/cancel store boundary를 만든다. 검증: invalid transition과 idempotency duplicate를 테스트한다. -- [ ] [op-pick] worker가 pending operation을 안전하게 pick하는 기준을 만든다. 검증: 같은 operation이 중복 pick되지 않는다. +- [x] [op-store] operation create/start/succeed/fail/cancel store boundary를 만든다. 검증: invalid transition과 idempotency duplicate를 테스트한다. +- [x] [op-pick] worker가 pending operation을 안전하게 pick하는 기준을 만든다. 검증: 같은 operation이 중복 pick되지 않는다. ### Epic: [outbox] Durable events operation 결과를 durable event로 남기고 fanout 대상이 읽을 수 있게 한다. -- [ ] [event-model] normalized event type과 payload 저장 형식을 정리한다. +- [x] [event-model] normalized event type과 payload 저장 형식을 정리한다. - [ ] [event-store] event append/list/mark-published boundary를 만든다. 검증: event order와 retry 가능한 publish 상태를 확인한다. -- [ ] [protosocket-events] proto-socket `event` channel 후보와 outbox event mapping을 contracts에 기록한다. +- [x] [protosocket-events] proto-socket `event` channel 후보와 outbox event mapping을 contracts에 기록한다. ## 완료 리뷰 diff --git a/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_0.log b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_0.log new file mode 100644 index 0000000..131ba13 --- /dev/null +++ b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_0.log @@ -0,0 +1,203 @@ + + +# Code Review Reference - API + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record the needed decision in `사용자 리뷰 요청` and stop for code-review. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-06-14 +task=m-operation-event-outbox/03+01_operation_event_store, plan=0, tag=API + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- Task ids: + - `event-store`: event append/list/mark-published boundary를 만든다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-cloud-G05.md` → `code_review_cloud_G05_N.log`, `PLAN-local-G06.md` → `plan_local_G06_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-operation-event-outbox/03+01_operation_event_store/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. `USER_REVIEW.md`가 사용자 결정으로 완료/PASS 해소되면 code-review가 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log` 작성 후 archive 이동한다. +4. PASS이고 task group이 `m-operation-event-outbox`이면 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [API-1] Wire Postgres OperationEventStore | [x] | +| [API-2] Persist, List, and Mark Published Operation Events | [x] | + +## 구현 체크리스트 + +- [x] `PgStore`가 `pgOperationEventStore`를 초기화하고 `OperationEvents()`에서 반환한다. +- [x] `OperationEventStore`가 append/list/pending-list/mark-published boundary를 표현하도록 interface와 fakes를 갱신한다. +- [x] migration이 `published_at`을 저장하고 `operation_events` ordering/pending 조회에 필요한 index를 제공한다. +- [x] `AppendEvent`가 operation id, event id/type/subject, JSON object payload를 검증한 뒤 `operation_events`에 저장한다. +- [x] `ListEvents`, `ListPendingEvents`, `MarkPublished`가 payload와 publish 상태를 복원한다. 검증: event order와 retry 가능한 publish 상태를 확인한다. +- [x] `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`를 추가해 non-nil store, append/list round-trip, pending list, mark-published, invalid input, ordering을 검증한다. +- [x] `cd services/core && go test -count=1 ./internal/storage`와 `cd services/core && go test -count=1 ./...`를 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G05_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_local_G06_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [ ] PASS이면 active task 디렉터리 `agent-task/m-operation-event-outbox/03+01_operation_event_store/`를 `agent-task/archive/YYYY/MM/m-operation-event-outbox/03+01_operation_event_store/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [ ] PASS이고 task group이 `m-operation-event-outbox`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-operation-event-outbox/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-local-G06.md`와 `CODE_REVIEW-cloud-G05.md`를 작성하고 `complete.log`를 작성하지 않는다. +- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. +- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +- `services/core/internal/storage/postgres.go`에서는 `events` 패키지 import가 필요하지 않아 `events` import를 추가했다가 컴파일 에러("imported and not used")가 발생하여 제외했습니다. +- 반면 `services/core/internal/storage/postgres_test.go`에서는 `events.Event` 및 이벤트 관련 상수들을 직접 사용하여 컴파일 에러("undefined: events")가 발생했기 때문에 `"git.toki-labs.com/toki/gito/services/core/internal/events"` import를 추가하였습니다. + +## 주요 설계 결정 + +- `AppendEvent` payload 검증 시 `json.Unmarshal`을 이용하여 dynamic map(`map[string]any`)으로 파싱 가능 여부를 확인했습니다. unmarshaled map이 `nil`인 경우(예: 입력 JSON이 `null`인 경우) `ErrInvalidOperation`을 반환하여 JSON object만 영속화되도록 제한했습니다. +- `operation_events` 테이블 조회 속도 향상 및 미발행 이벤트 추출 최적화를 위해 partial index `ix_operation_events_pending_ordering`를 `WHERE published_at IS NULL` 조건으로 생성했습니다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- `PgStore.OperationEvents()`가 nil을 반환하지 않고 concrete store를 반환하는지 확인한다. +- `OperationEventStore` interface 변경이 storage fake 구현과 compile-time checks까지 반영됐는지 확인한다. +- payload validation이 JSON object만 허용하고 nil/empty payload를 `{}`로 정규화하는지 확인한다. +- `ListEvents` ordering이 `(created_at, id)`이고 operation id filter를 빠뜨리지 않았는지 확인한다. +- `ListPendingEvents`와 `MarkPublished`가 `published_at` 기준으로 재시도 가능한 pending 상태를 표현하는지 확인한다. +- Postgres integration test가 env 미설정 시 기존 guard로 SKIP되는지, env 설정 시 append/list behavior를 검증하는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. +- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다. + +### API-1 중간 검증 +```text +$ cd services/core && go test -count=1 ./internal/storage +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.004s +``` + +### API-2 중간 검증 +```text +$ cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +=== RUN TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates + postgres_test.go:731: GITO_TEST_DATABASE_URL not set; skipping Postgres integration test +--- SKIP: TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates (0.00s) +PASS +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.003s +``` + +### 최종 검증 +```text +$ git diff --check +(no output) + +$ cd services/core && go test -count=1 ./internal/storage +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.004s + +$ cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +=== RUN TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates + postgres_test.go:731: GITO_TEST_DATABASE_URL not set; skipping Postgres integration test +--- SKIP: TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates (0.00s) +PASS +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.003s + +$ cd services/core && go test -count=1 ./... +? git.toki-labs.com/toki/gito/services/core/cmd/server [no test files] +? git.toki-labs.com/toki/gito/services/core/cmd/shell [no test files] +? git.toki-labs.com/toki/gito/services/core/cmd/worker [no test files] +? git.toki-labs.com/toki/gito/services/core/internal/agentshell [no test files] +ok git.toki-labs.com/toki/gito/services/core/internal/config 0.002s +ok git.toki-labs.com/toki/gito/services/core/internal/controlplane 0.245s +ok git.toki-labs.com/toki/gito/services/core/internal/core 0.002s +? git.toki-labs.com/toki/gito/services/core/internal/events [no test files] +ok git.toki-labs.com/toki/gito/services/core/internal/gitengine 0.832s +ok git.toki-labs.com/toki/gito/services/core/internal/protosocket 0.004s +? git.toki-labs.com/toki/gito/services/core/internal/provider [no test files] +ok git.toki-labs.com/toki/gito/services/core/internal/provider/forgejo 0.002s +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.004s +ok git.toki-labs.com/toki/gito/services/core/internal/worker 0.003s +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +Sections and their ownership: + +| Section | Owner | Note | +|---------|-------|------| +| Header comment, 개요, 리뷰 에이전트 지시 | Fixed at stub creation | Implementing agent must not modify or execute these (archive, complete.log, and task-directory archive move are review-agent only) | +| Roadmap Targets | Fixed at stub creation from plan when present | Implementing agent must not modify; code-review copies it into `complete.log` as `Roadmap Completion` only on PASS | +| 구현 항목별 완료 여부 (item names) | Fixed at stub creation | Implementing agent checks `[ ]` -> `[x]` only | +| 구현 체크리스트 (item text/order) | Fixed at stub creation from plan | Implementing agent checks `[ ]` -> `[x]` only; final checkbox is mandatory before saving | +| 코드리뷰 전용 체크리스트 | Review agent only | Implementing agent must not modify or check this section | +| 계획 대비 변경 사항, 주요 설계 결정 | Implementing agent | Replace placeholder text with actual content | +| 사용자 리뷰 요청 | Implementing agent | Keep `상태: 없음` unless user input is required to proceed; do not ask the user directly during implementation; when filled, include exact decision, evidence, commands/output, why automatic follow-up cannot resolve it, and resume condition | +| 리뷰어를 위한 체크포인트 | Fixed at stub creation | Pre-filled from plan | +| 검증 결과 (section headings + commands) | Fixed at stub creation | Implementing agent fills in command output only; command changes require a `계획 대비 변경 사항` entry | +| 코드리뷰 결과 | Review agent appends | Not included in stub | + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - correctness: Pass + - completeness: Fail + - test coverage: Fail + - API contract: Warn + - code quality: Pass + - plan deviation: Fail + - verification trust: Warn +- 발견된 문제: + - Required: `services/core/migrations/00001_initial.sql:51` only adds a pending partial index. `ListEvents` filters by `operation_id` and orders by `(created_at, id)`, so the planned "ordering/pending 조회에 필요한 index" is incomplete. Add `CREATE INDEX IF NOT EXISTS ix_operation_events_operation_ordering ON operation_events (operation_id, created_at, id)` and update the migration snippet test to require it. + - Required: `services/core/internal/storage/postgres_test.go:836` compares a Postgres round-tripped `TIMESTAMPTZ` directly against `time.Now().UTC()`. PostgreSQL stores microsecond precision, so a real `GITO_TEST_DATABASE_URL` run can fail when `now` has nanoseconds. Truncate the expected value to microseconds or compare against `now.Truncate(time.Microsecond)`. + - Required: `services/core/internal/storage/postgres_test.go:806` verifies only count/order for `ListEvents`. The plan required payload JSONB round-trip, operation id/type/subject/created_at restoration, and invalid missing type/subject plus non-object payload coverage. Extend `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates` to assert those fields and boundary cases. +- 다음 단계: FAIL 후속 `PLAN-local-G06.md`와 `CODE_REVIEW-cloud-G05.md`를 작성한다. diff --git a/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_1.log b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_1.log new file mode 100644 index 0000000..a2708c2 --- /dev/null +++ b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_1.log @@ -0,0 +1,194 @@ + + +# Code Review Reference - REVIEW_API + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record the needed decision in `사용자 리뷰 요청` and stop for code-review. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-06-14 +task=m-operation-event-outbox/03+01_operation_event_store, plan=1, tag=REVIEW_API + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- Task ids: + - `event-store`: event append/list/mark-published boundary를 만든다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-cloud-G05.md` -> `code_review_cloud_G05_N.log`, `PLAN-local-G06.md` -> `plan_local_G06_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-operation-event-outbox/03+01_operation_event_store/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. `USER_REVIEW.md`가 사용자 결정으로 완료/PASS 해소되면 code-review가 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log` 작성 후 archive 이동한다. +4. PASS이고 task group이 `m-operation-event-outbox`이면 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_API-1] Add Operation Event List Ordering Index | [x] | +| [REVIEW_API-2] Strengthen Operation Event Store Integration Assertions | [x] | + +## 구현 체크리스트 + +- [x] `operation_events` migration에 operation-specific ordering index를 추가하고 migration snippet test가 이를 확인한다. +- [x] `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`가 timestamp precision, payload/metadata round-trip, invalid type/subject/non-object payload boundary를 검증한다. +- [x] `git diff --check`, `cd services/core && go test -count=1 ./internal/storage`, focused Postgres test, `cd services/core && go test -count=1 ./...`를 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G05_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_local_G06_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-operation-event-outbox/03+01_operation_event_store/`를 `agent-task/archive/YYYY/MM/m-operation-event-outbox/03+01_operation_event_store/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-operation-event-outbox`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-operation-event-outbox/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-local-G06.md`와 `CODE_REVIEW-cloud-G05.md`를 작성하고 `complete.log`를 작성하지 않는다. +- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. +- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +- 없음. 계획에 있는 모든 지시사항(Operation Event List Ordering Index 추가, migration snippet test 반영, integration test의 timestamp precision/round-trip/invalid input/empty payload normalization 추가 및 json/reflect 패키지 import)을 정확하게 이행했습니다. + +## 주요 설계 결정 + +- `assertJSONPayload` helper 함수를 추가하여 byte array로 전달되는 dynamic JSON payload를 unmarshal 후 `reflect.DeepEqual`을 통해 의미론적(semantic) 동등성으로 검증하도록 설계했습니다. +- Postgres `TIMESTAMPTZ` 필드가 microsecond precision을 가지기 때문에 integration test에서 timestamp assertions 시 `time.Now().UTC().Truncate(time.Microsecond)`로 microsecond 단위까지 Truncate하여 DB 및 integration test의 값 정합성을 맞추었습니다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- `operation_events`에 `(operation_id, created_at, id)` list ordering index가 추가됐는지 확인한다. +- migration snippet test가 pending index와 operation ordering index를 모두 요구하는지 확인한다. +- focused Postgres test가 timestamp 값을 microsecond precision으로 비교하는지 확인한다. +- `ListEvents` assertions가 operation id, event id/type/subject, payload, created_at, published_at 상태를 검증하는지 확인한다. +- invalid input coverage가 missing type/subject와 JSON string/null payload를 포함하는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. +- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다. + +### REVIEW_API-1 중간 검증 +```text +$ cd services/core && go test -count=1 ./internal/storage +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.005s +``` + +### REVIEW_API-2 중간 검증 +```text +$ cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +=== RUN TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates + postgres_test.go:747: GITO_TEST_DATABASE_URL not set; skipping Postgres integration test +--- SKIP: TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates (0.00s) +PASS +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.004s +``` + +### 최종 검증 +```text +$ git diff --check +(no output) + +$ cd services/core && go test -count=1 ./internal/storage +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.005s + +$ cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +=== RUN TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates + postgres_test.go:747: GITO_TEST_DATABASE_URL not set; skipping Postgres integration test +--- SKIP: TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates (0.00s) +PASS +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.004s + +$ cd services/core && go test -count=1 ./... +? git.toki-labs.com/toki/gito/services/core/cmd/server [no test files] +? git.toki-labs.com/toki/gito/services/core/cmd/shell [no test files] +? git.toki-labs.com/toki/gito/services/core/cmd/worker [no test files] +? git.toki-labs.com/toki/gito/services/core/internal/agentshell [no test files] +ok git.toki-labs.com/toki/gito/services/core/internal/config 0.003s +ok git.toki-labs.com/toki/gito/services/core/internal/controlplane 0.369s +ok git.toki-labs.com/toki/gito/services/core/internal/core 0.003s +? git.toki-labs.com/toki/gito/services/core/internal/events [no test files] +ok git.toki-labs.com/toki/gito/services/core/internal/gitengine 1.174s +ok git.toki-labs.com/toki/gito/services/core/internal/protosocket 0.005s +? git.toki-labs.com/toki/gito/services/core/internal/provider [no test files] +ok git.toki-labs.com/toki/gito/services/core/internal/provider/forgejo 0.003s +ok git.toki-labs.com/toki/gito/services/core/internal/storage 0.006s +ok git.toki-labs.com/toki/gito/services/core/internal/worker 0.004s +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +Sections and their ownership: + +| Section | Owner | Note | +|---------|-------|------| +| Header comment, 개요, 리뷰 에이전트 지시 | Fixed at stub creation | Implementing agent must not modify or execute these (archive, complete.log, and task-directory archive move are review-agent only) | +| Roadmap Targets | Fixed at stub creation from plan when present | Implementing agent must not modify; code-review copies it into `complete.log` as `Roadmap Completion` only on PASS | +| 구현 항목별 완료 여부 (item names) | Fixed at stub creation | Implementing agent checks `[ ]` -> `[x]` only | +| 구현 체크리스트 (item text/order) | Fixed at stub creation from plan | Implementing agent checks `[ ]` -> `[x]` only; final checkbox is mandatory before saving | +| 코드리뷰 전용 체크리스트 | Review agent only | Implementing agent must not modify or check this section | +| 계획 대비 변경 사항, 주요 설계 결정 | Implementing agent | Replace placeholder text with actual content | +| 사용자 리뷰 요청 | Implementing agent | Keep `상태: 없음` unless user input is required to proceed; do not ask the user directly during implementation; when filled, include exact decision, evidence, commands/output, why automatic follow-up cannot resolve it, and resume condition | +| 리뷰어를 위한 체크포인트 | Fixed at stub creation | Pre-filled from plan | +| 검증 결과 (section headings + commands) | Fixed at stub creation | Implementing agent fills in command output only; command changes require a `계획 대비 변경 사항` entry | +| 코드리뷰 결과 | Review agent appends | Not included in stub | + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS 완료 처리로 `complete.log`를 작성하고 task directory를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/complete.log b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/complete.log new file mode 100644 index 0000000..4d889e4 --- /dev/null +++ b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/complete.log @@ -0,0 +1,44 @@ +# Complete - m-operation-event-outbox/03+01_operation_event_store + +## 완료 일시 + +2026-06-13 + +## 요약 + +Operation event store boundary 구현과 review follow-up을 2회 리뷰 루프로 완료했다. Final verdict: PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_local_G06_0.log` | `code_review_cloud_G05_0.log` | FAIL | operation event list ordering index와 Postgres integration test precision/round-trip coverage 보완 필요. | +| `plan_local_G06_1.log` | `code_review_cloud_G05_1.log` | PASS | operation-specific ordering index, migration snippet check, timestamp precision, payload/metadata round-trip, invalid payload/field coverage 보강 완료. | + +## 구현/정리 내용 + +- `PgStore.OperationEvents()`가 Postgres-backed `OperationEventStore`를 반환하도록 연결하고 append/list/pending-list/mark-published boundary를 구현했다. +- `operation_events`에 `published_at`, operation-specific ordering index, pending partial index를 추가했다. +- Postgres integration test가 event metadata, JSON payload round-trip, pending 상태, mark-published timestamp, invalid input, nil payload normalization을 검증하도록 보강했다. + +## 최종 검증 + +- `git diff --check` - PASS; no output. +- `cd services/core && go test -count=1 ./internal/storage` - PASS; `ok git.toki-labs.com/toki/gito/services/core/internal/storage`. +- `cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage` - PASS with expected SKIP because `GITO_TEST_DATABASE_URL` is unset; output includes `GITO_TEST_DATABASE_URL not set; skipping Postgres integration test`. +- `cd services/core && go test -count=1 ./...` - PASS; all packages passed or reported `[no test files]`. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- Completed task ids: + - `event-store`: PASS; evidence=`plan_local_G06_0.log`, `code_review_cloud_G05_0.log`, `plan_local_G06_1.log`, `code_review_cloud_G05_1.log`; verification=`git diff --check`, `cd services/core && go test -count=1 ./internal/storage`, `cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage`, `cd services/core && go test -count=1 ./...` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_0.log b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_0.log new file mode 100644 index 0000000..ac4a6de --- /dev/null +++ b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_0.log @@ -0,0 +1,295 @@ + + +# Plan - Operation Event Store + +## 이 파일을 읽는 구현 에이전트에게 + +`CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채우는 것이 구현의 필수 마지막 단계다. 구현 후 검증 명령을 실행하고 실제 출력, 계획 대비 변경, 주요 설계 결정을 기록한 뒤 active 파일을 그대로 두고 리뷰 준비를 보고한다. 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 범위 충돌 없이는 안전하게 진행할 수 없으면 review stub의 `사용자 리뷰 요청`에 근거를 기록하고 중단한다. 구현 중 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 말고, `USER_REVIEW.md`, `complete.log`, archive 이동은 만들지 않는다. 후속 에이전트가 명령 재실행이나 증거 수집으로 해소할 수 있는 검증 공백은 사용자 리뷰 요청이 아니다. + +## 배경 + +현재 Operation Event Outbox Milestone의 outbox Epic에서 `event-model`과 `protosocket-events`는 바로 정리되었고, 남은 큰 항목은 durable `operation_events` store 구현이다. Migration에는 `operation_events` table이 있지만 `PgStore.OperationEvents()`는 아직 `nil`을 반환하고, publish 완료 상태도 schema/interface에 없다. 이 plan은 `OperationEventStore`를 append/list/mark-published boundary로 확장해 Postgres 구현과 검증을 만든다. + +## 사용자 리뷰 요청 흐름 + +구현 중 차단 조건은 active review stub의 `사용자 리뷰 요청` 섹션에 기록한다. 해당 섹션은 `agent-ops/skills/common/_templates/implementation-user-review-request-section.md` 형식을 따른다. 구현 에이전트는 직접 사용자에게 묻지 않고, code-review가 사용자 리뷰 필요성을 검증하고 `USER_REVIEW.md` 작성 여부를 결정한다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- Task ids: + - `event-store`: event append/list/mark-published boundary를 만든다. +- Completion mode: check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-roadmap/current.md` +- `agent-roadmap/phase/control-plane-foundation/PHASE.md` +- `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- `agent-test/local/rules.md` +- `agent-test/local/core-smoke.md` +- `agent-test/local/contracts-smoke.md` +- `services/core/go.mod` +- `services/core/internal/events/events.go` +- `services/core/internal/core/model.go` +- `services/core/internal/storage/storage.go` +- `services/core/internal/storage/postgres.go` +- `services/core/migrations/00001_initial.sql` +- `services/core/internal/storage/storage_test.go` +- `services/core/internal/storage/postgres_test.go` +- `services/core/internal/controlplane/runtime_test.go` +- `packages/contracts/notes/gito-control-plane.md` +- `agent-task/archive/2026/06/m-operation-event-outbox/01_operation_store/complete.log` + +### 테스트 환경 규칙 + +- `test_env`: local. +- `agent-test/local/rules.md` 존재, 읽음. core 변경은 `agent-test/local/core-smoke.md`로 라우팅된다. +- 적용 명령: core 코드 변경 최소 `cd services/core && go test ./...`; storage behavior 변경은 해당 동작 테스트 또는 차단 사유 필요. +- `contracts-smoke`도 읽었지만 이 plan은 contract 파일을 추가로 수정하지 않는다. 작은 contract 정리는 이미 별도 직접 변경으로 처리되었다. +- Postgres integration tests는 기존 `GITO_TEST_DATABASE_URL` guard가 없으면 SKIP된다. 구현 에이전트는 skip을 실제 출력으로 남기고, 가능한 경우 같은 명령을 DB env가 설정된 runner에서 실행한다. + +### 테스트 커버리지 공백 + +- `operation_events` table 존재 여부는 `storage_test.go` migration snippet test가 확인한다. +- `PgStore.OperationEvents()` non-nil, append/list, payload JSONB round-trip, `(created_at, id)` ordering은 아직 테스트가 없다. +- `mark-published` 상태는 현재 schema/interface에 없다. 이번 plan에서 `published_at`과 pending list/mark published API를 함께 추가해야 `event-store` Task를 PASS 때 체크할 수 있다. + +### 심볼 참조 + +- renamed/removed symbol: none. +- `OperationEvents()` references: `services/core/internal/storage/postgres.go:64`, `services/core/internal/storage/storage.go:18`, `services/core/internal/storage/storage.go:166`, `services/core/internal/storage/storage_test.go:155`, `services/core/internal/controlplane/runtime_test.go:43`. +- `OperationEventStore` references: `services/core/internal/storage/storage.go:80`, `services/core/internal/storage/storage_test.go:86`. +- `operation_events` schema references: `services/core/migrations/00001_initial.sql:41`, `services/core/internal/storage/storage_test.go:190`. + +### 분할 판단 + +- split decision policy evaluated before file creation. +- shared task group: `agent-task/m-operation-event-outbox/`. +- this subtask: `03+01_operation_event_store`, depends on predecessor index `01`. +- predecessor `01_operation_store` satisfied by archived `agent-task/archive/2026/06/m-operation-event-outbox/01_operation_store/complete.log`. +- sibling `02+01_operation_pick` is complete but not a dependency for this implementation. +- 내부 추가 split은 하지 않는다. schema/interface/Postgres/test가 함께 움직여야 하는 하나의 storage boundary이며, 나누면 중간 상태가 compile되지 않는다. + +### 범위 결정 근거 + +- Include: `services/core/internal/storage/storage.go`, `services/core/internal/storage/postgres.go`, `services/core/migrations/00001_initial.sql`, `services/core/internal/storage/storage_test.go`, `services/core/internal/storage/postgres_test.go`, and fake store compile updates in `services/core/internal/controlplane/runtime_test.go`. +- Include migration shape change for `published_at`, because `mark-published` is part of the roadmap Task. +- Exclude: proto-socket `event.ack`, durable consumer cursor, Redis fanout, runtime auto-append, and provider branch event persistence. Those require separate routing beyond the current `OperationEventStore` boundary. + +### 빌드 등급 + +- build lane: `local-G06`; bounded storage implementation with deterministic unit/integration tests, but includes Postgres JSONB and FK behavior. +- review lane: `cloud-G05`; storage boundary and integration-test skip semantics need careful review. + +## 구현 체크리스트 + +- [ ] `PgStore`가 `pgOperationEventStore`를 초기화하고 `OperationEvents()`에서 반환한다. +- [ ] `OperationEventStore`가 append/list/pending-list/mark-published boundary를 표현하도록 interface와 fakes를 갱신한다. +- [ ] migration이 `published_at`을 저장하고 `operation_events` ordering/pending 조회에 필요한 index를 제공한다. +- [ ] `AppendEvent`가 operation id, event id/type/subject, JSON object payload를 검증한 뒤 `operation_events`에 저장한다. +- [ ] `ListEvents`, `ListPendingEvents`, `MarkPublished`가 payload와 publish 상태를 복원한다. 검증: event order와 retry 가능한 publish 상태를 확인한다. +- [ ] `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`를 추가해 non-nil store, append/list round-trip, pending list, mark-published, invalid input, ordering을 검증한다. +- [ ] `cd services/core && go test -count=1 ./internal/storage`와 `cd services/core && go test -count=1 ./...`를 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [API-1] Wire Postgres OperationEventStore + +#### 문제 + +[postgres.go](/config/workspace/gito/services/core/internal/storage/postgres.go:19)의 `PgStore`는 repos/leases/ops/watches/cursors/deliveries만 보유하고, [postgres.go](/config/workspace/gito/services/core/internal/storage/postgres.go:64)의 `OperationEvents()`는 `nil`을 반환한다. 반면 [storage.go](/config/workspace/gito/services/core/internal/storage/storage.go:80)는 operation event store interface를 이미 제공하지만 `mark-published` boundary가 없다. + +Before: + +```go +// services/core/internal/storage/postgres.go:19 +type PgStore struct { + pool *pgxpool.Pool + repos *pgRepoStore + leases *pgWorkspaceLeaseStore + ops *pgOperationStore + watches *pgBranchWatchStore + cursors *pgRevisionCursorStore + deliveries *pgProviderDeliveryStore +} + +// services/core/internal/storage/postgres.go:61 +func (s *PgStore) Repos() RepoStore { return s.repos } +func (s *PgStore) WorkspaceLeases() WorkspaceLeaseStore { return s.leases } +func (s *PgStore) Operations() OperationStore { return s.ops } +func (s *PgStore) OperationEvents() OperationEventStore { return nil } +func (s *PgStore) BranchWatches() BranchWatchStore { return s.watches } +func (s *PgStore) RevisionCursors() RevisionCursorStore { return s.cursors } +func (s *PgStore) ProviderDeliveries() ProviderDeliveryStore { return s.deliveries } +``` + +#### 해결 방법 + +`storage.OperationEvent`에 `PublishedAt *time.Time`을 추가하고 `OperationEventStore`에 `ListPendingEvents(ctx, limit int)`와 `MarkPublished(ctx, eventID string, now time.Time)`를 추가한다. `PgStore`에 `opEvents *pgOperationEventStore`를 추가하고 `NewPgStore`에서 초기화한다. `encoding/json`과 `events` import를 추가한다. `pgOperationEventStore`는 `operation_events` table을 사용한다. + +After shape: + +```go +type PgStore struct { + pool *pgxpool.Pool + repos *pgRepoStore + leases *pgWorkspaceLeaseStore + ops *pgOperationStore + opEvents *pgOperationEventStore + watches *pgBranchWatchStore + cursors *pgRevisionCursorStore + deliveries *pgProviderDeliveryStore +} + +s.opEvents = &pgOperationEventStore{pool: pool} + +func (s *PgStore) OperationEvents() OperationEventStore { return s.opEvents } +``` + +#### 수정 파일 및 체크리스트 + +- [ ] `services/core/internal/storage/postgres.go`: imports에 `encoding/json`과 `events`를 추가한다. +- [ ] `services/core/internal/storage/storage.go`: `OperationEvent.PublishedAt`, `ListPendingEvents`, `MarkPublished`를 추가한다. +- [ ] `services/core/internal/storage/postgres.go`: `PgStore` struct와 `NewPgStore` 초기화에 `opEvents`를 추가한다. +- [ ] `services/core/internal/storage/postgres.go`: `OperationEvents()`가 `s.opEvents`를 반환하게 한다. +- [ ] `services/core/internal/storage/postgres.go`: `pgOperationEventStore`, `normalizeOperationEvent`, `scanOperationEvent` helper를 추가한다. +- [ ] `services/core/internal/storage/storage_test.go`와 `services/core/internal/controlplane/runtime_test.go`: fake `OperationEventStore` 구현을 새 interface에 맞춘다. + +#### 테스트 작성 + +`services/core/internal/storage/postgres_test.go`에 `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`를 추가한다. + +#### 중간 검증 + +```bash +cd services/core && go test -count=1 ./internal/storage +``` + +Expected: storage package passes; if `GITO_TEST_DATABASE_URL` is absent, Postgres integration tests print existing SKIP reasons. + +### [API-2] Persist, List, and Mark Published Operation Events + +#### 문제 + +[00001_initial.sql](/config/workspace/gito/services/core/migrations/00001_initial.sql:41)는 `operation_events` table을 만들지만 publish 상태 컬럼이 없다. [storage.go](/config/workspace/gito/services/core/internal/storage/storage.go:75)의 `OperationEvent`는 `events.Event` payload를 들고 있어 JSONB round-trip, ordering, pending retry 상태를 구현해야 한다. + +Before: + +```go +// services/core/internal/storage/storage.go:75 +type OperationEvent struct { + OperationID string + Event events.Event +} + +type OperationEventStore interface { + AppendEvent(ctx context.Context, ev OperationEvent) error + ListEvents(ctx context.Context, operationID string) ([]OperationEvent, error) +} +``` + +#### 해결 방법 + +`AppendEvent` validates: + +- `OperationID`, `Event.ID`, `Event.Type`, `Event.Subject` are non-empty. +- nil/empty payload becomes `{}`. +- non-empty payload must be valid JSON object, not array/string/null. +- `CreatedAt` zero becomes `time.Now().UTC()`, otherwise UTC normalized. +- `PublishedAt` must be nil on append; publish completion is controlled by `MarkPublished`. + +SQL: + +```sql +INSERT INTO operation_events (id, operation_id, type, subject, payload, created_at) +VALUES ($1, $2, $3, $4, $5, $6) +``` + +`ListEvents` trims `operationID`, rejects empty input with `ErrInvalidOperation`, and runs: + +```sql +SELECT operation_id, id, type, subject, payload, created_at, published_at +FROM operation_events +WHERE operation_id=$1 +ORDER BY created_at, id +``` + +`ListPendingEvents` rejects non-positive limits and returns unpublished rows: + +```sql +SELECT operation_id, id, type, subject, payload, created_at, published_at +FROM operation_events +WHERE published_at IS NULL +ORDER BY created_at, id +LIMIT $1 +``` + +`MarkPublished` sets `published_at` and returns the marked event: + +```sql +UPDATE operation_events +SET published_at=$2 +WHERE id=$1 +RETURNING operation_id, id, type, subject, payload, created_at, published_at +``` + +#### 수정 파일 및 체크리스트 + +- [ ] `services/core/migrations/00001_initial.sql`: `published_at TIMESTAMPTZ`와 pending/order index를 추가한다. +- [ ] `services/core/internal/storage/storage_test.go`: migration snippet test가 `published_at TIMESTAMPTZ`와 index를 확인하게 한다. +- [ ] `services/core/internal/storage/postgres.go`: append SQL을 추가하고 FK/unique errors are wrapped as `append operation event`. +- [ ] `services/core/internal/storage/postgres.go`: list/pending/mark-published SQL을 추가하고 rows scanning errors are wrapped as `scan operation event`. +- [ ] `services/core/internal/storage/postgres_test.go`: repo와 operation을 seed한 뒤 two events를 append한다. +- [ ] `services/core/internal/storage/postgres_test.go`: listed events의 order, operation id, type, subject, payload bytes, created_at을 assert한다. +- [ ] `services/core/internal/storage/postgres_test.go`: pending list가 unpublished events만 반환하고 `MarkPublished` 뒤에는 해당 event가 빠지는지 assert한다. +- [ ] `services/core/internal/storage/postgres_test.go`: empty operation id, missing event id/type/subject, invalid JSON array/string payload, non-positive pending limit, missing mark id return `ErrInvalidOperation` or `ErrOperationNotFound` as appropriate. + +#### 테스트 작성 + +Write `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates` in `services/core/internal/storage/postgres_test.go`. Assertions: + +- `store.OperationEvents()` is non-nil. +- append two events with payloads `{"state":"running"}` and `{"state":"succeeded"}`. +- list returns two events ordered by `(created_at, id)`. +- pending list returns both before publish, then only the second after `MarkPublished(first.ID, now)`. +- marked event has non-nil `PublishedAt`. +- invalid payload `[]byte("[1]")` is rejected. +- invalid empty fields are rejected. + +#### 중간 검증 + +```bash +cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +``` + +Expected: PASS when `GITO_TEST_DATABASE_URL` is set; existing SKIP is acceptable only when the env var is absent and must be recorded verbatim. + +## 의존 관계 및 구현 순서 + +The directory name `03+01_operation_event_store` means predecessor index `01` must be complete before implementation. This is satisfied by `agent-task/archive/2026/06/m-operation-event-outbox/01_operation_store/complete.log`. Implement API-1 before API-2 because tests need the concrete store to be wired. + +## 수정 파일 요약 + +| File | Items | +| --- | --- | +| `services/core/internal/storage/storage.go` | API-1 | +| `services/core/internal/storage/postgres.go` | API-1, API-2 | +| `services/core/migrations/00001_initial.sql` | API-2 | +| `services/core/internal/storage/storage_test.go` | API-1, API-2 | +| `services/core/internal/storage/postgres_test.go` | API-1, API-2 | +| `services/core/internal/controlplane/runtime_test.go` | API-1 | + +## 최종 검증 + +```bash +git diff --check +cd services/core && go test -count=1 ./internal/storage +cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +cd services/core && go test -count=1 ./... +``` + +Expected: `git diff --check` has no output. Go tests pass; the focused Postgres test may SKIP only when `GITO_TEST_DATABASE_URL` is unset, and that skip must be pasted into the review stub. + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_1.log b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_1.log new file mode 100644 index 0000000..d567d1d --- /dev/null +++ b/agent-task/archive/2026/06/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_1.log @@ -0,0 +1,224 @@ + + +# Plan - Operation Event Store Review Follow-up + +## 이 파일을 읽는 구현 에이전트에게 + +`CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채우는 것이 구현의 필수 마지막 단계다. 구현 후 검증 명령을 실행하고 실제 출력, 계획 대비 변경, 주요 설계 결정을 기록한 뒤 active 파일을 그대로 두고 리뷰 준비를 보고한다. 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 범위 충돌 없이는 안전하게 진행할 수 없으면 review stub의 `사용자 리뷰 요청`에 근거를 기록하고 중단한다. 구현 중 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 말고, `USER_REVIEW.md`, `complete.log`, archive 이동은 만들지 않는다. 후속 에이전트가 명령 재실행이나 증거 수집으로 해소할 수 있는 검증 공백은 사용자 리뷰 요청이 아니다. + +## 배경 + +첫 리뷰에서 `OperationEventStore` 구현 방향은 맞지만, 계획상 필수였던 operation-specific ordering index와 Postgres 통합 테스트 assertion이 부족해 `FAIL` 판정을 받았다. 이 follow-up은 production API를 넓히지 않고 migration/test 보완만으로 `event-store` Task를 닫을 수 있게 만든다. + +## 사용자 리뷰 요청 흐름 + +구현 중 차단 조건은 active review stub의 `사용자 리뷰 요청` 섹션에 기록한다. 해당 섹션은 `agent-ops/skills/common/_templates/implementation-user-review-request-section.md` 형식을 따른다. 구현 에이전트는 직접 사용자에게 묻지 않고, code-review가 사용자 리뷰 필요성을 검증하고 `USER_REVIEW.md` 작성 여부를 결정한다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- Task ids: + - `event-store`: event append/list/mark-published boundary를 만든다. +- Completion mode: check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-roadmap/current.md` +- `agent-roadmap/phase/control-plane-foundation/PHASE.md` +- `agent-roadmap/phase/control-plane-foundation/milestones/operation-event-outbox.md` +- `agent-test/local/rules.md` +- `agent-test/local/core-smoke.md` +- `agent-ops/rules/project/domain/core/rules.md` +- `agent-ops/rules/project/domain/contracts/rules.md` +- `services/core/internal/storage/storage.go` +- `services/core/internal/storage/postgres.go` +- `services/core/migrations/00001_initial.sql` +- `services/core/internal/storage/storage_test.go` +- `services/core/internal/storage/postgres_test.go` +- `services/core/internal/controlplane/runtime_test.go` +- `services/core/internal/events/events.go` +- `packages/contracts/notes/gito-control-plane.md` +- `agent-task/m-operation-event-outbox/03+01_operation_event_store/plan_local_G06_0.log` +- `agent-task/m-operation-event-outbox/03+01_operation_event_store/code_review_cloud_G05_0.log` +- `agent-task/archive/2026/06/m-operation-event-outbox/01_operation_store/complete.log` + +### 테스트 환경 규칙 + +- `test_env`: local. +- `agent-test/local/rules.md` 존재, 읽음. +- matched profile: `agent-test/local/core-smoke.md`, 읽음. +- core 변경 최소 검증은 `cd services/core && go test ./...`다. +- storage behavior 변경은 해당 동작 테스트가 필요하다. +- fresh execution이 필요하므로 Go 검증은 `-count=1`을 사용한다. +- Postgres integration test는 `GITO_TEST_DATABASE_URL`이 없으면 SKIP될 수 있지만, env가 있을 때 통과하도록 timestamp precision과 assertion을 맞춰야 한다. + +### 테스트 커버리지 공백 + +- `operation_events` pending index는 테스트되지만, `ListEvents(operationID)`의 `(operation_id, created_at, id)` ordering index는 migration과 snippet test에 없다. +- `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`는 event count/order/pending/mark 일부만 확인하고, operation id/type/subject/payload/created_at round-trip을 확인하지 않는다. +- 같은 테스트가 Postgres `TIMESTAMPTZ` microsecond precision을 고려하지 않고 `PublishedAt`을 `time.Now().UTC()`와 직접 비교한다. +- invalid input coverage가 empty operation id/event id, array payload, non-positive limit, missing mark id에 그치며 missing type/subject, string/null payload가 빠져 있다. + +### 심볼 참조 + +- renamed/removed symbol: none. +- `OperationEvents()` references: `services/core/internal/storage/postgres.go`, `services/core/internal/storage/storage.go`, `services/core/internal/storage/storage_test.go`, `services/core/internal/controlplane/runtime_test.go`, `services/core/internal/storage/postgres_test.go`. +- `ListPendingEvents` / `MarkPublished` references: `services/core/internal/storage/storage.go`, `services/core/internal/storage/postgres.go`, `services/core/internal/storage/storage_test.go`, `services/core/internal/storage/postgres_test.go`. +- `operation_events` schema references: `services/core/migrations/00001_initial.sql`, `services/core/internal/storage/storage_test.go`, `services/core/internal/storage/postgres.go`, `services/core/internal/storage/postgres_test.go`. + +### 분할 판단 + +- split decision policy evaluated before writing this follow-up. +- 기존 split task directory `03+01_operation_event_store`를 유지한다. follow-up은 같은 storage boundary의 review-required 보완이며 새 sibling으로 나누면 같은 active task의 FAIL 해소 추적이 흩어진다. +- predecessor index `01`은 `agent-task/archive/2026/06/m-operation-event-outbox/01_operation_store/complete.log`로 충족됐다. + +### 범위 결정 근거 + +- Include: `services/core/migrations/00001_initial.sql`, `services/core/internal/storage/storage_test.go`, `services/core/internal/storage/postgres_test.go`. +- Include production `services/core/internal/storage/postgres.go` only if strengthened tests expose an actual implementation defect. +- Exclude: proto-socket event runtime, durable consumer cursor, Redis fanout, roadmap 문서 갱신, contracts note 추가 변경. + +### 빌드 등급 + +- build lane: `local-G06`; bounded storage/migration/test follow-up with deterministic commands. +- review lane: `cloud-G05`; Postgres schema and integration-test skip/precision semantics need careful review. + +## 구현 체크리스트 + +- [ ] `operation_events` migration에 operation-specific ordering index를 추가하고 migration snippet test가 이를 확인한다. +- [ ] `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`가 timestamp precision, payload/metadata round-trip, invalid type/subject/non-object payload boundary를 검증한다. +- [ ] `git diff --check`, `cd services/core && go test -count=1 ./internal/storage`, focused Postgres test, `cd services/core && go test -count=1 ./...`를 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [REVIEW_API-1] Add Operation Event List Ordering Index + +#### 문제 + +`services/core/internal/storage/postgres.go:938`의 `ListEvents`는 `WHERE operation_id=$1 ORDER BY created_at, id`로 조회한다. 하지만 `services/core/migrations/00001_initial.sql:51`에는 pending partial index만 있어 계획의 "ordering/pending 조회에 필요한 index"가 완성되지 않았다. + +Before: + +```sql +CREATE INDEX IF NOT EXISTS ix_operation_events_pending_ordering + ON operation_events (created_at, id) + WHERE published_at IS NULL; +``` + +#### 해결 방법 + +operation별 list query를 위해 `(operation_id, created_at, id)` index를 추가하고, migration snippet test가 해당 index 이름을 요구하게 한다. + +After shape: + +```sql +CREATE INDEX IF NOT EXISTS ix_operation_events_operation_ordering + ON operation_events (operation_id, created_at, id); + +CREATE INDEX IF NOT EXISTS ix_operation_events_pending_ordering + ON operation_events (created_at, id) + WHERE published_at IS NULL; +``` + +#### 수정 파일 및 체크리스트 + +- [ ] `services/core/migrations/00001_initial.sql`: `ix_operation_events_operation_ordering` index를 추가한다. +- [ ] `services/core/internal/storage/storage_test.go`: `TestInitialMigrationSupportsStoreBoundary`의 required snippets에 새 index 이름을 추가한다. + +#### 테스트 작성 + +`services/core/internal/storage/storage_test.go`의 기존 migration snippet test를 갱신한다. 새 별도 테스트는 만들지 않는다. + +#### 중간 검증 + +```bash +cd services/core && go test -count=1 ./internal/storage +``` + +Expected: storage package passes. + +### [REVIEW_API-2] Strengthen Operation Event Store Integration Assertions + +#### 문제 + +`services/core/internal/storage/postgres_test.go:836`은 Postgres round-trip timestamp를 nanosecond precision의 `time.Now().UTC()`와 직접 비교한다. `TIMESTAMPTZ`는 microsecond precision이므로 DB env가 설정된 runner에서 실패할 수 있다. 또한 `services/core/internal/storage/postgres_test.go:806` 이후의 assertions는 operation id/type/subject/payload/created_at round-trip과 missing type/subject, string/null payload boundary를 확인하지 않는다. + +Before: + +```go +now := time.Now().UTC() +marked, err := eventStore.MarkPublished(ctx, "evt-001", now) +if err != nil { + t.Fatalf("mark published: %v", err) +} +if marked.PublishedAt == nil { + t.Fatal("expected non-nil PublishedAt after mark published") +} +if !marked.PublishedAt.Equal(now) { + t.Fatalf("expected PublishedAt to be %v, got %v", now, *marked.PublishedAt) +} +``` + +#### 해결 방법 + +Seed timestamps and publish timestamp with `Truncate(time.Microsecond)` before writing them, or compare returned timestamps to truncated expected values. Add semantic JSON payload comparison helper inside the test file and assert listed events restore `OperationID`, `Event.ID`, `Event.Type`, `Event.Subject`, `Event.Payload`, `Event.CreatedAt`, and `PublishedAt`. Extend invalid cases for missing type, missing subject, JSON string payload, and JSON null payload. Also assert nil/empty payload normalization to `{}` if a third event is added for that boundary. + +After shape: + +```go +created1 := time.Now().UTC().Truncate(time.Microsecond) +created2 := created1.Add(time.Second) +now := time.Now().UTC().Truncate(time.Microsecond) + +if !listed[0].Event.CreatedAt.Equal(created1) { + t.Fatalf("created_at round trip: got %v want %v", listed[0].Event.CreatedAt, created1) +} +assertJSONPayload(t, listed[0].Event.Payload, `{"state":"running"}`) +``` + +#### 수정 파일 및 체크리스트 + +- [ ] `services/core/internal/storage/postgres_test.go`: seeded `CreatedAt` and publish `now` values use microsecond precision. +- [ ] `services/core/internal/storage/postgres_test.go`: `ListEvents` assertions cover operation id, event id/type/subject, JSON payload, created_at, and nil `PublishedAt` before mark. +- [ ] `services/core/internal/storage/postgres_test.go`: `MarkPublished` assertion compares against microsecond-truncated expected value. +- [ ] `services/core/internal/storage/postgres_test.go`: invalid cases cover empty type, empty subject, JSON string payload, and JSON null payload. +- [ ] `services/core/internal/storage/postgres_test.go`: if helper imports are needed, add only standard-library imports such as `encoding/json` and `reflect`. + +#### 테스트 작성 + +Extend `TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates`; do not create a separate test unless the helper makes the existing test too large to read. + +#### 중간 검증 + +```bash +cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +``` + +Expected: PASS with `GITO_TEST_DATABASE_URL`; SKIP is acceptable only when the env var is absent and the exact skip output is recorded. + +## 의존 관계 및 구현 순서 + +Apply `REVIEW_API-1` before `REVIEW_API-2` so the focused integration test and migration snippet check reflect the final schema. Predecessor split index `01` is already satisfied by `agent-task/archive/2026/06/m-operation-event-outbox/01_operation_store/complete.log`. + +## 수정 파일 요약 + +| File | Items | +| --- | --- | +| `services/core/migrations/00001_initial.sql` | REVIEW_API-1 | +| `services/core/internal/storage/storage_test.go` | REVIEW_API-1 | +| `services/core/internal/storage/postgres_test.go` | REVIEW_API-2 | +| `services/core/internal/storage/postgres.go` | REVIEW_API-2 only if strengthened tests expose a production defect | + +## 최종 검증 + +```bash +git diff --check +cd services/core && go test -count=1 ./internal/storage +cd services/core && go test -count=1 -run TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates -v ./internal/storage +cd services/core && go test -count=1 ./... +``` + +Expected: `git diff --check` has no output. Go tests pass. The focused Postgres test may SKIP only when `GITO_TEST_DATABASE_URL` is unset, and that skip must be pasted into the review stub. + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/packages/contracts/notes/gito-control-plane.md b/packages/contracts/notes/gito-control-plane.md index 52daa54..d05806f 100644 --- a/packages/contracts/notes/gito-control-plane.md +++ b/packages/contracts/notes/gito-control-plane.md @@ -114,6 +114,16 @@ platform features. ## Normalized Events +Durable event records use a stable envelope: + +| Field | Meaning | +| --- | --- | +| `id` | Stable event id generated by Gito. | +| `type` | Normalized event name from the table below. | +| `subject` | Entity anchor such as `operation:` or `repo:`. | +| `payload` | JSON object encoded as `application/json`; raw provider secrets and credential values are forbidden. | +| `created_at` | Time Gito recorded the event. | + | Event | Meaning | | --- | --- | | `repo.changed` | Repo registry changed. | @@ -130,3 +140,34 @@ platform features. | `change_request.updated` | PR/MR-like object changed. | | `change_request.merged` | PR/MR-like object merged. | | `provider.webhook.received` | Provider webhook received. | + +### Operation Event Payloads + +Operation lifecycle events are stored in the outbox as normalized event records +with `subject=operation:`. + +| Event | Required payload fields | +| --- | --- | +| `operation.started` | `operation_id`, `repo_id`, `type`, `state`, `started_at` | +| `operation.completed` | `operation_id`, `repo_id`, `type`, `state`, `completed_at` | +| `operation.failed` | `operation_id`, `repo_id`, `type`, `state`, `failed_at`, optional `error` | + +### Outbox Storage Contract + +- PostgreSQL is the source of truth for durable events. +- Stored payloads are JSON objects and map to `events.Event.Payload` with + `events.PayloadEncodingJSON`. +- Outbox readers list events in `(created_at, id)` order. +- Publish retry state belongs to the outbox store boundary; Redis, if added + later, only accelerates fanout. + +### proto-socket Event Mapping + +- Stored `type` maps to proto-socket `action` on the `event` channel. +- Stored `payload` maps to the proto-socket envelope `payload` object without + provider secrets or credential values. +- `event.subscribe` filters by event name and optional payload fields such as + `repo_id` and `branch`. +- `event.list` returns recent event payloads for smoke/debug use. +- `event.ack` remains placeholder-only until a durable cursor or delivery + acknowledgement Milestone explicitly scopes it. diff --git a/services/core/internal/events/events.go b/services/core/internal/events/events.go index 106778e..7b695cd 100644 --- a/services/core/internal/events/events.go +++ b/services/core/internal/events/events.go @@ -3,9 +3,10 @@ package events import "time" type Event struct { - ID string - Type string - Subject string + ID string + Type string + Subject string + // Payload is stored as a JSON object in the durable outbox. Payload []byte CreatedAt time.Time } @@ -26,3 +27,5 @@ const ( ChangeRequestMerged = "change_request.merged" ProviderWebhook = "provider.webhook.received" ) + +const PayloadEncodingJSON = "application/json" diff --git a/services/core/internal/storage/postgres.go b/services/core/internal/storage/postgres.go index 18ab0f2..17a1bb8 100644 --- a/services/core/internal/storage/postgres.go +++ b/services/core/internal/storage/postgres.go @@ -3,6 +3,7 @@ package storage import ( "context" "database/sql" + "encoding/json" "errors" "fmt" "strings" @@ -21,6 +22,7 @@ type PgStore struct { repos *pgRepoStore leases *pgWorkspaceLeaseStore ops *pgOperationStore + opEvents *pgOperationEventStore watches *pgBranchWatchStore cursors *pgRevisionCursorStore deliveries *pgProviderDeliveryStore @@ -48,6 +50,7 @@ func NewPgStore(ctx context.Context, dsn string, migrationSQL string) (*PgStore, s.repos = &pgRepoStore{pool: pool} s.leases = &pgWorkspaceLeaseStore{pool: pool} s.ops = &pgOperationStore{pool: pool} + s.opEvents = &pgOperationEventStore{pool: pool} s.watches = &pgBranchWatchStore{pool: pool} s.cursors = &pgRevisionCursorStore{pool: pool} s.deliveries = &pgProviderDeliveryStore{pool: pool} @@ -61,7 +64,7 @@ func (s *PgStore) Ping(ctx context.Context) error { return s.pool.Ping(ctx) } func (s *PgStore) Repos() RepoStore { return s.repos } func (s *PgStore) WorkspaceLeases() WorkspaceLeaseStore { return s.leases } func (s *PgStore) Operations() OperationStore { return s.ops } -func (s *PgStore) OperationEvents() OperationEventStore { return nil } +func (s *PgStore) OperationEvents() OperationEventStore { return s.opEvents } func (s *PgStore) BranchWatches() BranchWatchStore { return s.watches } func (s *PgStore) RevisionCursors() RevisionCursorStore { return s.cursors } func (s *PgStore) ProviderDeliveries() ProviderDeliveryStore { return s.deliveries } @@ -868,3 +871,168 @@ func uniqueViolationConstraint(err error) (string, bool) { } return "", false } + +// pgOperationEventStore + +type pgOperationEventStore struct { + pool *pgxpool.Pool +} + +func (s *pgOperationEventStore) AppendEvent(ctx context.Context, ev OperationEvent) error { + opID := strings.TrimSpace(ev.OperationID) + eventID := strings.TrimSpace(ev.Event.ID) + eventType := strings.TrimSpace(ev.Event.Type) + subject := strings.TrimSpace(ev.Event.Subject) + + if opID == "" || eventID == "" || eventType == "" || subject == "" { + return fmt.Errorf("%w: operation_id, event_id, type, and subject are required", ErrInvalidOperation) + } + + if ev.PublishedAt != nil { + return fmt.Errorf("%w: published_at must be nil on append", ErrInvalidOperation) + } + + payload, err := validateAndNormalizePayload(ev.Event.Payload) + if err != nil { + return err + } + + createdAt := ev.Event.CreatedAt + if createdAt.IsZero() { + createdAt = time.Now().UTC() + } else { + createdAt = createdAt.UTC() + } + + const q = ` + INSERT INTO operation_events (id, operation_id, type, subject, payload, created_at) + VALUES ($1, $2, $3, $4, $5, $6) + ` + _, err = s.pool.Exec(ctx, q, eventID, opID, eventType, subject, payload, createdAt) + if err != nil { + return fmt.Errorf("append operation event: %w", err) + } + return nil +} + +func validateAndNormalizePayload(payload []byte) ([]byte, error) { + if len(payload) == 0 { + return []byte("{}"), nil + } + var m map[string]any + if err := json.Unmarshal(payload, &m); err != nil { + return nil, fmt.Errorf("%w: payload must be a valid JSON object", ErrInvalidOperation) + } + if m == nil { + return nil, fmt.Errorf("%w: payload cannot be null", ErrInvalidOperation) + } + return payload, nil +} + +func (s *pgOperationEventStore) ListEvents(ctx context.Context, operationID string) ([]OperationEvent, error) { + operationID = strings.TrimSpace(operationID) + if operationID == "" { + return nil, fmt.Errorf("%w: operation_id is required", ErrInvalidOperation) + } + + const q = ` + SELECT operation_id, id, type, subject, payload, created_at, published_at + FROM operation_events + WHERE operation_id=$1 + ORDER BY created_at, id + ` + rows, err := s.pool.Query(ctx, q, operationID) + if err != nil { + return nil, fmt.Errorf("list operation events: %w", err) + } + defer rows.Close() + + var out []OperationEvent + for rows.Next() { + ev, err := scanOperationEvent(rows) + if err != nil { + return nil, fmt.Errorf("scan operation event: %w", err) + } + out = append(out, ev) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list operation events rows error: %w", err) + } + return out, nil +} + +func (s *pgOperationEventStore) ListPendingEvents(ctx context.Context, limit int) ([]OperationEvent, error) { + if limit <= 0 { + return nil, fmt.Errorf("%w: limit must be positive", ErrInvalidOperation) + } + + const q = ` + SELECT operation_id, id, type, subject, payload, created_at, published_at + FROM operation_events + WHERE published_at IS NULL + ORDER BY created_at, id + LIMIT $1 + ` + rows, err := s.pool.Query(ctx, q, limit) + if err != nil { + return nil, fmt.Errorf("list pending operation events: %w", err) + } + defer rows.Close() + + var out []OperationEvent + for rows.Next() { + ev, err := scanOperationEvent(rows) + if err != nil { + return nil, fmt.Errorf("scan operation event: %w", err) + } + out = append(out, ev) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list pending operation events rows error: %w", err) + } + return out, nil +} + +func (s *pgOperationEventStore) MarkPublished(ctx context.Context, eventID string, now time.Time) (OperationEvent, error) { + eventID = strings.TrimSpace(eventID) + if eventID == "" { + return OperationEvent{}, fmt.Errorf("%w: event_id is required", ErrInvalidOperation) + } + + const q = ` + UPDATE operation_events + SET published_at=$2 + WHERE id=$1 + RETURNING operation_id, id, type, subject, payload, created_at, published_at + ` + ev, err := scanOperationEvent(s.pool.QueryRow(ctx, q, eventID, now.UTC())) + if err != nil { + if isNoRows(err) { + return OperationEvent{}, fmt.Errorf("%w: event %s not found", ErrOperationNotFound, eventID) + } + return OperationEvent{}, fmt.Errorf("scan operation event: %w", err) + } + return ev, nil +} + +func scanOperationEvent(scanner repoScanner) (OperationEvent, error) { + var ev OperationEvent + var publishedAt sql.NullTime + if err := scanner.Scan( + &ev.OperationID, + &ev.Event.ID, + &ev.Event.Type, + &ev.Event.Subject, + &ev.Event.Payload, + &ev.Event.CreatedAt, + &publishedAt, + ); err != nil { + return OperationEvent{}, err + } + ev.Event.CreatedAt = ev.Event.CreatedAt.UTC() + if publishedAt.Valid { + t := publishedAt.Time.UTC() + ev.PublishedAt = &t + } + return ev, nil +} diff --git a/services/core/internal/storage/postgres_test.go b/services/core/internal/storage/postgres_test.go index f50e7a6..69eb177 100644 --- a/services/core/internal/storage/postgres_test.go +++ b/services/core/internal/storage/postgres_test.go @@ -2,10 +2,12 @@ package storage_test import ( "context" + "encoding/json" "errors" "fmt" "net/url" "os" + "reflect" "strconv" "strings" "sync" @@ -15,6 +17,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "git.toki-labs.com/toki/gito/services/core/internal/core" + "git.toki-labs.com/toki/gito/services/core/internal/events" "git.toki-labs.com/toki/gito/services/core/internal/storage" ) @@ -723,3 +726,262 @@ func TestPostgresOperationStorePickQueuedOperationDoesNotDuplicate(t *testing.T) t.Errorf("success picked count: got %d want %d", successCount, seedCount) } } + +func assertJSONPayload(t *testing.T, actual []byte, expected string) { + t.Helper() + var actMap, expMap map[string]any + if err := json.Unmarshal(actual, &actMap); err != nil { + t.Fatalf("unmarshal actual payload %q: %v", string(actual), err) + } + if err := json.Unmarshal([]byte(expected), &expMap); err != nil { + t.Fatalf("unmarshal expected payload %q: %v", expected, err) + } + if !reflect.DeepEqual(actMap, expMap) { + t.Errorf("payload mismatch:\ngot: %+v\nwant: %+v", actMap, expMap) + } +} + +func TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates(t *testing.T) { + dsn := os.Getenv("GITO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test") + } + + ctx := context.Background() + dsn = isolatedPostgresDSN(t, ctx, dsn) + + migrationData, err := os.ReadFile("../../migrations/00001_initial.sql") + if err != nil { + t.Fatalf("read migration: %v", err) + } + + store, err := storage.NewPgStore(ctx, dsn, string(migrationData)) + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + + if store.OperationEvents() == nil { + t.Fatal("OperationEvents() must not return nil") + } + + repo := core.Repo{ + ID: "event-repo", + Name: "Event Repo", + RemoteURL: "ssh://git.example.invalid/toki/event-repo.git", + DefaultBranch: "main", + WorkspaceRoot: "/workspaces/event-repo", + } + if err := store.Repos().CreateRepo(ctx, repo); err != nil { + t.Fatalf("create repo: %v", err) + } + + op := core.Operation{ + ID: "op-event-001", + RepoID: repo.ID, + Type: core.OperationClone, + State: core.OperationQueued, + IdempotencyKey: "idem-event-1", + CreatedBy: "test-event-appender", + } + if _, err := store.Operations().CreateOperation(ctx, op); err != nil { + t.Fatalf("create operation: %v", err) + } + + eventStore := store.OperationEvents() + + created1 := time.Now().UTC().Truncate(time.Microsecond) + created2 := created1.Add(time.Second) + + // Append two events + ev1 := storage.OperationEvent{ + OperationID: op.ID, + Event: events.Event{ + ID: "evt-001", + Type: events.OperationStarted, + Subject: "clone", + Payload: []byte(`{"state":"running"}`), + CreatedAt: created1, + }, + } + ev2 := storage.OperationEvent{ + OperationID: op.ID, + Event: events.Event{ + ID: "evt-002", + Type: events.OperationCompleted, + Subject: "clone", + Payload: []byte(`{"state":"succeeded"}`), + CreatedAt: created2, + }, + } + + if err := eventStore.AppendEvent(ctx, ev1); err != nil { + t.Fatalf("append event 1: %v", err) + } + if err := eventStore.AppendEvent(ctx, ev2); err != nil { + t.Fatalf("append event 2: %v", err) + } + + // Append third event to test empty/nil payload normalization to {} + ev3 := storage.OperationEvent{ + OperationID: op.ID, + Event: events.Event{ + ID: "evt-003", + Type: events.OperationCompleted, + Subject: "clone", + Payload: nil, + CreatedAt: created2.Add(time.Second), + }, + } + if err := eventStore.AppendEvent(ctx, ev3); err != nil { + t.Fatalf("append event 3 (nil payload): %v", err) + } + + // List events and check ordering + listed, err := eventStore.ListEvents(ctx, op.ID) + if err != nil { + t.Fatalf("list events: %v", err) + } + if len(listed) != 3 { + t.Fatalf("expected 3 listed events, got %d", len(listed)) + } + + // Verify details of each event (Round-trip) + // Event 1 + if listed[0].OperationID != op.ID { + t.Errorf("ev1: operationID mismatch: got %q want %q", listed[0].OperationID, op.ID) + } + if listed[0].Event.ID != "evt-001" { + t.Errorf("ev1: eventID mismatch: got %q want %q", listed[0].Event.ID, "evt-001") + } + if listed[0].Event.Type != events.OperationStarted { + t.Errorf("ev1: type mismatch: got %q want %q", listed[0].Event.Type, events.OperationStarted) + } + if listed[0].Event.Subject != "clone" { + t.Errorf("ev1: subject mismatch: got %q want %q", listed[0].Event.Subject, "clone") + } + assertJSONPayload(t, listed[0].Event.Payload, `{"state":"running"}`) + if !listed[0].Event.CreatedAt.Equal(created1) { + t.Errorf("ev1: CreatedAt mismatch: got %v want %v", listed[0].Event.CreatedAt, created1) + } + if listed[0].PublishedAt != nil { + t.Errorf("ev1: PublishedAt must be nil initially") + } + + // Event 2 + if listed[1].OperationID != op.ID { + t.Errorf("ev2: operationID mismatch") + } + if listed[1].Event.ID != "evt-002" { + t.Errorf("ev2: eventID mismatch") + } + assertJSONPayload(t, listed[1].Event.Payload, `{"state":"succeeded"}`) + if !listed[1].Event.CreatedAt.Equal(created2) { + t.Errorf("ev2: CreatedAt mismatch") + } + + // Event 3 (nil payload normalization check) + if listed[2].Event.ID != "evt-003" { + t.Errorf("ev3: eventID mismatch") + } + assertJSONPayload(t, listed[2].Event.Payload, `{}`) + + // List pending events before publish + pending, err := eventStore.ListPendingEvents(ctx, 10) + if err != nil { + t.Fatalf("list pending: %v", err) + } + if len(pending) != 3 { + t.Fatalf("expected 3 pending events, got %d", len(pending)) + } + + // Mark first event as published + now := time.Now().UTC().Truncate(time.Microsecond) + marked, err := eventStore.MarkPublished(ctx, "evt-001", now) + if err != nil { + t.Fatalf("mark published: %v", err) + } + if marked.PublishedAt == nil { + t.Fatal("expected non-nil PublishedAt after mark published") + } + if !marked.PublishedAt.Equal(now) { + t.Fatalf("expected PublishedAt to be %v, got %v", now, *marked.PublishedAt) + } + + // List pending events after publish + pendingAfter, err := eventStore.ListPendingEvents(ctx, 10) + if err != nil { + t.Fatalf("list pending after: %v", err) + } + if len(pendingAfter) != 2 { + t.Fatalf("expected 2 pending events, got %d", len(pendingAfter)) + } + if pendingAfter[0].Event.ID != "evt-002" || pendingAfter[1].Event.ID != "evt-003" { + t.Fatalf("unexpected pending events after publish") + } + + // Invalid input coverage + // 1. Invalid payload (array) + badPayloadEv := storage.OperationEvent{ + OperationID: op.ID, + Event: events.Event{ + ID: "evt-bad", + Type: events.OperationStarted, + Subject: "clone", + Payload: []byte(`[1]`), + CreatedAt: time.Now().UTC(), + }, + } + if err := eventStore.AppendEvent(ctx, badPayloadEv); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for invalid JSON array payload, got %v", err) + } + + // 2. Invalid payload (string) + badPayloadEv2 := badPayloadEv + badPayloadEv2.Event.Payload = []byte(`"abc"`) + if err := eventStore.AppendEvent(ctx, badPayloadEv2); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for invalid JSON string payload, got %v", err) + } + + // 3. Invalid payload (null) + badPayloadEv3 := badPayloadEv + badPayloadEv3.Event.Payload = []byte(`null`) + if err := eventStore.AppendEvent(ctx, badPayloadEv3); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for invalid JSON null payload, got %v", err) + } + + // 4. Missing fields + emptyOpIDEv := ev1 + emptyOpIDEv.OperationID = "" + if err := eventStore.AppendEvent(ctx, emptyOpIDEv); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for empty operation ID, got %v", err) + } + + emptyEventIDEv := ev1 + emptyEventIDEv.Event.ID = "" + if err := eventStore.AppendEvent(ctx, emptyEventIDEv); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for empty event ID, got %v", err) + } + + emptyTypeEv := ev1 + emptyTypeEv.Event.Type = "" + if err := eventStore.AppendEvent(ctx, emptyTypeEv); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for empty type, got %v", err) + } + + emptySubjectEv := ev1 + emptySubjectEv.Event.Subject = "" + if err := eventStore.AppendEvent(ctx, emptySubjectEv); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for empty subject, got %v", err) + } + + // 5. Non-positive limit for pending list + if _, err := eventStore.ListPendingEvents(ctx, 0); !errors.Is(err, storage.ErrInvalidOperation) { + t.Fatalf("expected ErrInvalidOperation for non-positive limit, got %v", err) + } + + // 6. Missing mark ID + if _, err := eventStore.MarkPublished(ctx, "evt-missing", time.Now()); !errors.Is(err, storage.ErrOperationNotFound) { + t.Fatalf("expected ErrOperationNotFound for missing mark ID, got %v", err) + } +} diff --git a/services/core/internal/storage/storage.go b/services/core/internal/storage/storage.go index d880054..db39b14 100644 --- a/services/core/internal/storage/storage.go +++ b/services/core/internal/storage/storage.go @@ -75,11 +75,14 @@ type OperationStore interface { type OperationEvent struct { OperationID string Event events.Event + PublishedAt *time.Time } type OperationEventStore interface { AppendEvent(ctx context.Context, ev OperationEvent) error ListEvents(ctx context.Context, operationID string) ([]OperationEvent, error) + ListPendingEvents(ctx context.Context, limit int) ([]OperationEvent, error) + MarkPublished(ctx context.Context, eventID string, now time.Time) (OperationEvent, error) } type BranchWatchStore interface { diff --git a/services/core/internal/storage/storage_test.go b/services/core/internal/storage/storage_test.go index 1e09664..2532b2f 100644 --- a/services/core/internal/storage/storage_test.go +++ b/services/core/internal/storage/storage_test.go @@ -82,6 +82,12 @@ func (f *fakeOpEvents) AppendEvent(_ context.Context, _ storage.OperationEvent) func (f *fakeOpEvents) ListEvents(_ context.Context, _ string) ([]storage.OperationEvent, error) { return nil, nil } +func (f *fakeOpEvents) ListPendingEvents(_ context.Context, _ int) ([]storage.OperationEvent, error) { + return nil, nil +} +func (f *fakeOpEvents) MarkPublished(_ context.Context, _ string, _ time.Time) (storage.OperationEvent, error) { + return storage.OperationEvent{}, nil +} var _ storage.OperationEventStore = (*fakeOpEvents)(nil) @@ -209,6 +215,9 @@ func TestInitialMigrationSupportsStoreBoundary(t *testing.T) { "ALTER TABLE revision_cursors DROP CONSTRAINT IF EXISTS revision_cursors_repo_id_fkey", "UNIQUE (provider, repo_id, branch)", "UNIQUE (provider, dedupe_key)", + "published_at TIMESTAMPTZ", + "CREATE INDEX IF NOT EXISTS ix_operation_events_operation_ordering", + "CREATE INDEX IF NOT EXISTS ix_operation_events_pending_ordering", } forbiddenSnippets := []string{ "CREATE TABLE IF NOT EXISTS revision_cursors (\n repo_id TEXT NOT NULL REFERENCES repos(id)", diff --git a/services/core/migrations/00001_initial.sql b/services/core/migrations/00001_initial.sql index 553d493..958e8d9 100644 --- a/services/core/migrations/00001_initial.sql +++ b/services/core/migrations/00001_initial.sql @@ -44,9 +44,17 @@ CREATE TABLE IF NOT EXISTS operation_events ( type TEXT NOT NULL, subject TEXT NOT NULL, payload JSONB NOT NULL DEFAULT '{}'::jsonb, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + published_at TIMESTAMPTZ ); +CREATE INDEX IF NOT EXISTS ix_operation_events_operation_ordering + ON operation_events (operation_id, created_at, id); + +CREATE INDEX IF NOT EXISTS ix_operation_events_pending_ordering + ON operation_events (created_at, id) + WHERE published_at IS NULL; + CREATE TABLE IF NOT EXISTS revision_cursors ( repo_id TEXT NOT NULL, branch TEXT NOT NULL,