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
This commit is contained in:
parent
3ed8797b28
commit
07b4e63366
13 changed files with 1463 additions and 9 deletions
|
|
@ -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에 기록한다.
|
||||
|
||||
## 완료 리뷰
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,203 @@
|
|||
<!-- task=m-operation-event-outbox/03+01_operation_event_store plan=0 tag=API -->
|
||||
|
||||
# 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`를 작성한다.
|
||||
|
|
@ -0,0 +1,194 @@
|
|||
<!-- task=m-operation-event-outbox/03+01_operation_event_store plan=1 tag=REVIEW_API -->
|
||||
|
||||
# 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로 이동한다.
|
||||
|
|
@ -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
|
||||
|
||||
- 없음
|
||||
|
||||
## 후속 작업
|
||||
|
||||
- 없음
|
||||
|
|
@ -0,0 +1,295 @@
|
|||
<!-- task=m-operation-event-outbox/03+01_operation_event_store plan=0 tag=API -->
|
||||
|
||||
# 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`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다.
|
||||
|
|
@ -0,0 +1,224 @@
|
|||
<!-- task=m-operation-event-outbox/03+01_operation_event_store plan=1 tag=REVIEW_API -->
|
||||
|
||||
# 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`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다.
|
||||
|
|
@ -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:<operation_id>` or `repo:<repo_id>`. |
|
||||
| `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:<operation_id>`.
|
||||
|
||||
| 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.
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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)",
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Reference in a new issue