feat: update project rules, roadmap, and core service changes
- Update agent-ops project rules and roadmap files - Add proto-socket infrastructure communication rail milestone - Update Flutter pubspec.lock and contracts notes - Enhance core service: config, HTTP middleware, router - Add notification module improvements - Add protosocket internal package
This commit is contained in:
parent
9551709b5f
commit
e36db7281b
28 changed files with 1836 additions and 80 deletions
|
|
@ -29,6 +29,7 @@ NomadCode는 AI 병렬 작업 운용을 위한 원레포다. 백엔드 오케스
|
|||
|
||||
- 원레포 루트에서 가능한 작업은 `bin/test`, `bin/lint`, `bin/build`, `bin/dev`를 우선 사용한다.
|
||||
- cross-boundary 변경은 계약, 서비스, 클라이언트 반영을 같은 브랜치에서 닫는다.
|
||||
- 내부 런타임 간 신규 통신 규약은 `proto-socket`을 기본 표준으로 삼는다. REST/HTTP는 health/readiness, 외부 provider 호환 API, 운영 smoke/curl, public callback, 단순 정적 조회처럼 `proto-socket`이 과한 경우에만 예외로 쓰고, 예외 사유를 contracts 또는 관련 domain rule에 남긴다.
|
||||
- `services/core` 변경은 `go test ./...` 또는 `bin/test`로 확인한다.
|
||||
- `apps/client` 변경은 `flutter test`와 필요 시 `flutter analyze --no-fatal-infos`로 확인한다.
|
||||
- `packages/contracts`가 실제 계약을 갖기 전에는 API 형태를 서비스/클라이언트에 중복 고정하지 말고 계약 후보를 문서화한다.
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@ NomadCode는 Flutter 기반 앱, core 서비스, 공유 계약, agent-operation
|
|||
- 요약: Flutter 앱에 섞인 Mattermost push notification migration을 분리했고, 현재 package 경로는 `../nexo/packages/messaging_flutter`다.
|
||||
- [진행중] Workflow Core
|
||||
- 경로: `agent-roadmap/phase/workflow-core/PHASE.md`
|
||||
- 요약: Roadmap Driven Agent-Ops Automation을 NomadCode Core와 MCP-first 제어 표면 방향으로 정리한 뒤 실제 e2e 흐름을 기준으로 task lifecycle, retry, timeout, notification event를 안정화한다.
|
||||
- 요약: Roadmap Driven Agent-Ops Automation과 proto-socket 내부 통신 레일을 정리한 뒤 client-core 통신과 실제 e2e 흐름을 기준으로 task lifecycle, retry, timeout, notification event를 안정화한다.
|
||||
- [진행중] External Integration
|
||||
- 경로: `agent-roadmap/phase/external-integration/PHASE.md`
|
||||
- 요약: Workflow Core 재개 전에 Mattermost/Nexo messaging 정합성을 먼저 닫고, 이후 Plane 확장, Mattermost, Agent Integrator, IOP OpenAI API Responses-compatible 호출을 실제 통합 흐름으로 확장한다.
|
||||
|
|
|
|||
|
|
@ -12,6 +12,9 @@
|
|||
- [진행중] Roadmap Driven Agent-Ops Automation
|
||||
- Phase: `agent-roadmap/phase/workflow-core/PHASE.md`
|
||||
- 경로: `agent-roadmap/phase/workflow-core/milestones/roadmap-driven-agent-ops-automation.md`
|
||||
- [진행중] Proto-Socket Infrastructure Communication Rail
|
||||
- Phase: `agent-roadmap/phase/workflow-core/PHASE.md`
|
||||
- 경로: `agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md`
|
||||
- [계획] Workflow Core
|
||||
- Phase: `agent-roadmap/phase/workflow-core/PHASE.md`
|
||||
- 경로: `agent-roadmap/phase/workflow-core/milestones/workflow-core.md`
|
||||
|
|
|
|||
|
|
@ -6,17 +6,21 @@
|
|||
|
||||
## 목표
|
||||
|
||||
Plane 통신 토대와 provider-neutral pipeline 설계 이후, 로드맵 기반 agent-ops 운영 루프를 NomadCode Core와 MCP-first 외부 제어 표면이 품는 방향으로 정리한다. 그 다음 실제 상태 변화를 기준으로 task lifecycle, enqueue/running/completed/failed 흐름을 안정화하고 retry, timeout, notification event의 최소 구조를 추가해 core workflow의 운영 기준을 만든다.
|
||||
Plane 통신 토대와 provider-neutral pipeline 설계 이후, 로드맵 기반 agent-ops 운영 루프를 NomadCode Core와 MCP-first 외부 제어 표면이 품는 방향으로 정리한다. 내부 런타임 간 신규 통신은 proto-socket 기본 레일로 통일하고, client-core 통신을 먼저 열어 실제 상태 변화를 기준으로 task lifecycle, enqueue/running/completed/failed 흐름을 안정화한다.
|
||||
|
||||
## Milestone 흐름
|
||||
|
||||
완료된 Milestone은 archive 경로를 가리키고, 검토중, 진행중, 계획, 스케치 또는 보류 Milestone은 이 Phase 하위 `milestones/` 경로를 가리킨다.
|
||||
완료, 검토중, 진행중, 계획, 스케치 순서를 기본으로 하되, 이번 Phase에서는 `Roadmap Driven Agent-Ops Automation`이 `Workflow Core` 구현의 선행 조건이므로 먼저 둔다.
|
||||
완료, 검토중, 진행중, 계획, 스케치 순서를 기본으로 하되, 이번 Phase에서는 `Roadmap Driven Agent-Ops Automation`과 `Proto-Socket Infrastructure Communication Rail`이 `Workflow Core` 구현의 선행 조건이므로 먼저 둔다.
|
||||
|
||||
- [진행중] Roadmap Driven Agent-Ops Automation
|
||||
- 경로: `agent-roadmap/phase/workflow-core/milestones/roadmap-driven-agent-ops-automation.md`
|
||||
- 요약: Workflow Core 구현 전에 사용자가 입력과 출력 검토를 NomadCode 라인에서 해결하고, 외부 에이전트는 MCP를 통해 roadmap/action core를 제어하는 운영 방향을 구현 가능한 계획으로 정리한다.
|
||||
|
||||
- [진행중] Proto-Socket Infrastructure Communication Rail
|
||||
- 경로: `agent-roadmap/phase/workflow-core/milestones/proto-socket-infrastructure-communication-rail.md`
|
||||
- 요약: 내부 런타임 간 신규 통신을 proto-socket 기본 레일로 통일하고, client-core task 요청과 상태 이벤트가 오갈 수 있는 기반을 만든다.
|
||||
|
||||
- [계획] Workflow Core
|
||||
- 경로: `agent-roadmap/phase/workflow-core/milestones/workflow-core.md`
|
||||
- 요약: Roadmap Driven Agent-Ops Automation의 Core/MCP-first 계약을 기준으로 task lifecycle, 상태 전이 책임, 실패 정책, retry/timeout, notification event 모델을 안정화한다.
|
||||
|
|
@ -25,5 +29,6 @@ Plane 통신 토대와 provider-neutral pipeline 설계 이후, 로드맵 기반
|
|||
|
||||
- Workflow Core는 canonical task lifecycle, retry/timeout envelope, heartbeat, terminal 상태 기록을 소유한다.
|
||||
- `Roadmap Driven Agent-Ops Automation`이 먼저 상태/책임/자동화 경계를 정리해야 `Workflow Core` 구현의 task lifecycle과 완료 이벤트 계약을 확정할 수 있다.
|
||||
- 내부 런타임 간 신규 통신은 `proto-socket`을 기본으로 하며, REST/HTTP 예외는 health/readiness, 외부 provider 호환 API, 운영 smoke/curl, public callback, 단순 정적 조회로 제한한다.
|
||||
- 외부 협업 도구 notification 발송, provider별 실제 comment/status 전송, 복잡한 workflow DSL은 후속 Phase로 넘긴다.
|
||||
- IOP/A2A/model runtime 내부 정책은 외부 실행 표면 또는 adapter 책임으로 둔다.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,93 @@
|
|||
# Milestone: Proto-Socket Infrastructure Communication Rail
|
||||
|
||||
## 위치
|
||||
|
||||
- Roadmap: `agent-roadmap/ROADMAP.md`
|
||||
- Phase: `agent-roadmap/phase/workflow-core/PHASE.md`
|
||||
|
||||
## 목표
|
||||
|
||||
NomadCode 내부 런타임 간 신규 통신을 `proto-socket` 중심으로 통일하는 기반을 만든다. 모든 내부 인프라 통신 후보를 대상으로 REST/HTTP 예외를 분리하고, 첫 구현 축은 Flutter client와 Core가 같은 레일로 task 조회, 실행 요청, 상태 이벤트를 주고받을 수 있게 한다.
|
||||
|
||||
## 상태
|
||||
|
||||
[진행중]
|
||||
|
||||
## 승격 조건
|
||||
|
||||
- 없음
|
||||
|
||||
## 구현 잠금
|
||||
|
||||
- 상태: 해제
|
||||
- 결정 필요: 없음
|
||||
|
||||
## 범위
|
||||
|
||||
- 내부 런타임 간 통신 표준을 `proto-socket` 기본, REST/HTTP 예외 방식으로 정리
|
||||
- `apps/client`와 `services/core` 사이의 client-core 통신 레일 구현 준비
|
||||
- `packages/contracts`에 proto-socket channel, message envelope, error/heartbeat/auth 후보를 정리
|
||||
- 기존 core REST task API를 proto-socket channel 후보로 매핑하고 전환 경계를 기록
|
||||
- Core의 proto-socket endpoint와 Flutter client connector의 최소 구현 단위 정의
|
||||
- health/readiness, 외부 provider 호환 API, 운영 smoke/curl, public callback, 단순 정적 조회 같은 REST/HTTP 예외 목록 관리
|
||||
|
||||
## 기능
|
||||
|
||||
### Epic: [protocol] Internal communication contract
|
||||
|
||||
내부 인프라 통신을 proto-socket 중심으로 통일하기 위한 계약과 예외 경계를 정리한다.
|
||||
|
||||
- [x] [standard-boundary] 내부 런타임 간 신규 통신은 `proto-socket`을 기본으로 삼고 REST/HTTP 예외 조건을 contracts 또는 domain rule에서 추적한다.
|
||||
- [x] [message-envelope] request, response, event, error, heartbeat, auth metadata를 포함한 최소 message envelope 후보를 정리한다.
|
||||
- [x] [rest-compat-map] 현재 core REST task API와 health/readiness endpoint를 proto-socket channel 후보와 REST 예외 목록으로 매핑한다.
|
||||
|
||||
### Epic: [core-rail] Core proto-socket rail
|
||||
|
||||
Core가 내부 런타임 통신용 proto-socket endpoint와 task channel을 제공할 수 있게 한다.
|
||||
|
||||
- [ ] [core-endpoint] `services/core`에 proto-socket endpoint 구성 위치와 lifecycle을 추가한다. 검증: Core가 기존 REST health/readiness를 유지하면서 proto-socket 연결 endpoint를 노출한다.
|
||||
- [ ] [task-channels] task create/list/get/enqueue와 task status event를 proto-socket channel로 제공한다. 검증: REST task API와 같은 의미의 요청/응답이 proto-socket 경로에서 동작한다.
|
||||
- [ ] [core-contract-tests] core proto-socket channel, auth, error, heartbeat 동작을 테스트로 고정한다.
|
||||
|
||||
### Epic: [client-rail] Flutter client-core rail
|
||||
|
||||
Flutter client가 no-op proto-socket facade를 실제 Core 연결 레일로 전환한다.
|
||||
|
||||
- [ ] [client-connector] `apps/client`의 `NoopProtoSocketConnector`를 대체할 concrete connector를 추가한다. 검증: 설정된 endpoint로 연결 상태가 `connected`까지 전이된다.
|
||||
- [ ] [client-task-api] workspace/task 화면이 mock 데이터 대신 Core proto-socket task channel을 소비할 수 있는 client service 경계를 만든다.
|
||||
- [ ] [client-tests] proto-socket endpoint 설정, reconnect/failure lifecycle, client task service를 Flutter test로 검증한다.
|
||||
|
||||
### Epic: [ops] Observability and rollout
|
||||
|
||||
내부 통신 레일을 운영 가능하게 만들고 예외 프로토콜이 다시 기본값으로 새지 않게 한다.
|
||||
|
||||
- [ ] [diagnostics] connection id, protocol version, channel name, error code를 로그와 디버그 표면에서 확인할 수 있게 한다.
|
||||
- [ ] [rollout-docs] 새 내부 통신을 추가할 때 proto-socket 우선 여부와 REST/HTTP 예외 사유를 확인하는 운영 체크를 문서화한다.
|
||||
|
||||
## 완료 리뷰
|
||||
|
||||
- 상태: 없음
|
||||
- 요청일: 없음
|
||||
- 완료 근거: 없음
|
||||
- 리뷰 필요:
|
||||
- [ ] 사용자가 완료 결과를 확인했다
|
||||
- [ ] archive 이동을 승인했다
|
||||
- 리뷰 코멘트: 없음
|
||||
|
||||
## 범위 제외
|
||||
|
||||
- Mattermost, Plane, Jira 같은 외부 provider의 native REST API 자체를 proto-socket으로 대체하지 않는다.
|
||||
- health/readiness, 운영 smoke/curl, public callback, 단순 정적 조회 REST 예외를 제거하지 않는다.
|
||||
- `../nexo/packages/messaging_flutter`의 native push 처리 로직을 수정하지 않는다.
|
||||
- IOP native protocol, A2A JSON-RPC, OpenAI-compatible Responses 같은 외부 실행 표면을 이 Milestone에서 재설계하지 않는다.
|
||||
- 전체 프로젝트 관리 UX 완성은 후속 Project Workspace Management UX 범위로 둔다.
|
||||
|
||||
## 작업 컨텍스트
|
||||
|
||||
- 관련 경로: `services/core/internal/http/`, `services/core/internal/workflow/`, `services/core/internal/scheduler/`, `apps/client/lib/src/integrations/proto_socket/`, `apps/client/lib/src/features/workspaces/`, `packages/contracts/`
|
||||
- 표준선(선택): 내부 런타임 간 신규 통신은 `proto-socket`을 기본 표준으로 삼고, REST/HTTP는 명시된 예외 조건에서만 유지한다.
|
||||
- 선행 작업: Mattermost Nexo Messaging Alignment, Client Integration Standardization, 프로젝트 룰의 proto-socket 표준 규약
|
||||
- 후속 작업: Workflow Core, External Integration, Project Workspace Management UX
|
||||
- 현재 지점: Nexo/Mattermost host 경계는 최신화와 테스트가 끝났고, 다음 작업은 client-core 통신이 가능하도록 내부 통신 레일을 까는 것이다.
|
||||
- 반영 근거: `packages/contracts/notes/flutter-core-api-candidates.md`에 내부 통신 원칙, proto-socket semantic envelope 후보, REST compatibility map 후보를 추가했다.
|
||||
- 확인 필요: 없음
|
||||
|
|
@ -26,6 +26,7 @@ Plane 통신 토대와 별도 pipeline 설계가 정리된 뒤 실제 상태 변
|
|||
- notification event 구조 정리
|
||||
- Plane 통신 토대와 후속 pipeline 설계에서 확인된 상태 변화와 실패 케이스 반영
|
||||
- Roadmap Driven Agent-Ops Automation의 Core/MCP-first 계약에서 정리된 로드맵/마일스톤/plan/review/완료 이벤트 책임 경계 반영
|
||||
- Proto-Socket Infrastructure Communication Rail에서 정리한 client-core 통신 레일과 내부 통신 예외 경계 반영
|
||||
|
||||
## 기능
|
||||
|
||||
|
|
@ -60,8 +61,8 @@ Core task lifecycle과 상태 전이 책임, 실패/재시도/timeout/notificati
|
|||
|
||||
- 이전 출처: `services/core/README.md`의 `## 단계별 다음 작업`
|
||||
- 주요 작업 영역: `services/core/internal/workflow/`, `services/core/internal/scheduler/`, `services/core/internal/notification/`
|
||||
- 선행 작업: Plane Communication Foundation, Work Item Provider Pipeline Design, Roadmap Driven Agent-Ops Automation
|
||||
- 선행 조건: Roadmap Driven Agent-Ops Automation이 `[계획]` 상태에서 Core/MCP-first 로드맵 action 계약을 기준선으로 제공한다.
|
||||
- 선행 작업: Plane Communication Foundation, Work Item Provider Pipeline Design, Roadmap Driven Agent-Ops Automation, Proto-Socket Infrastructure Communication Rail
|
||||
- 선행 조건: Roadmap Driven Agent-Ops Automation이 Core/MCP-first 로드맵 action 계약을 기준선으로 제공하고, Proto-Socket Infrastructure Communication Rail이 client-core task 요청과 상태 이벤트가 오갈 수 있는 내부 통신 기준선을 제공한다.
|
||||
- 후속 작업: External Integration
|
||||
- 현재 지점: Flutter-first Client Consolidation, Client Integration Standardization, Work Item Provider Pipeline Design이 완료되었고, workflow 상태 전이는 Roadmap Driven Agent-Ops Automation의 Core/MCP-first 방향성을 반영해 진행한다.
|
||||
- 책임 경계:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,80 @@
|
|||
# User Review Required - m-proto-socket-infrastructure-communication-rail/01_core_endpoint
|
||||
|
||||
## 요청 일시
|
||||
|
||||
2026-05-30
|
||||
|
||||
## 상태
|
||||
|
||||
RESOLVED
|
||||
|
||||
## 사유
|
||||
|
||||
- 유형: implementation-blocked
|
||||
- 현재 리뷰 회차: 3
|
||||
- 최종 판정: FAIL
|
||||
- 요약: proto-socket 원격 module path mismatch 때문에 plan의 필수 조건인 repo-external `replace` 제거를 저장소 안에서 안전하게 완료할 수 없다.
|
||||
|
||||
## 루프 이력
|
||||
|
||||
| Plan | Review | Verdict | 메모 |
|
||||
|------|--------|---------|------|
|
||||
| `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_0.log` | `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_0.log` | FAIL | repo-external `replace`, `retryable` 누락, authenticated route test 누락, gofmt 문제 |
|
||||
| `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_1.log` | `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_1.log` | FAIL | follow-up implementation이 placeholder 상태였고 이전 Required가 남음 |
|
||||
| `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_2.log` | `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_2.log` | FAIL | proto-socket module mismatch로 `replace` 제거 불가, 사용자 결정 필요 |
|
||||
|
||||
## 차단 근거
|
||||
|
||||
- 문제: `services/core/go.mod`가 아직 `git.toki-labs.com/toki/proto-socket/go v0.0.0-00010101000000-000000000000`와 `replace git.toki-labs.com/toki/proto-socket/go => ../../../proto-socket/go`에 의존한다.
|
||||
- 현재 archive plan: `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/plan_cloud_G08_2.log`
|
||||
- 현재 archive review: `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint/code_review_cloud_G08_2.log`
|
||||
- 검증 명령: `cd services/core && go mod tidy`
|
||||
- 실제 출력:
|
||||
|
||||
```text
|
||||
go: downloading git.toki-labs.com/toki/common-proto-socket/go v0.0.0-20260529210328-6f09e2d721c8
|
||||
go: found git.toki-labs.com/toki/common-proto-socket/go/packets in git.toki-labs.com/toki/common-proto-socket/go v0.0.0-20260529210328-6f09e2d721c8
|
||||
go: github.com/nomadcode/nomadcode-core/internal/protosocket imports
|
||||
git.toki-labs.com/toki/proto-socket/go imports
|
||||
git.toki-labs.com/toki/common-proto-socket/go/packets: git.toki-labs.com/toki/common-proto-socket/go@v0.0.0-20260529210328-6f09e2d721c8: parsing go.mod:
|
||||
module declares its path as: git.toki-labs.com/toki/proto-socket/go
|
||||
but was required as: git.toki-labs.com/toki/common-proto-socket/go
|
||||
```
|
||||
|
||||
- 차단 판단 근거: local `replace`를 커밋할지, 원격 proto-socket repository/module path를 먼저 고칠지, 또는 dependency 전략을 바꿀지는 repo 내부 구현자가 안전하게 단정할 수 없는 배포/의존성 정책 결정이다.
|
||||
|
||||
## 사용자 결정 필요
|
||||
|
||||
- [x] 자동 follow-up plan/review를 계속 진행한다.
|
||||
- [ ] 계획을 재작성한다.
|
||||
- [ ] 테스트 환경, secret, 외부 서비스, SDK, 장비 조건을 준비한 뒤 재시도한다.
|
||||
- [ ] 작업 범위를 줄이거나 보류/폐기한다.
|
||||
|
||||
## 해소 내용
|
||||
|
||||
- 사용자 결정: 원격 `proto-socket` 경로를 현재 환경과 맞춰 사용해도 된다는 승인.
|
||||
- 적용 결과: `services/core/go.mod`에서 repo-external `replace`를 제거하고, 원격 `git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a`로 고정했다.
|
||||
- 해소 검증:
|
||||
|
||||
```text
|
||||
$ GOPRIVATE=git.toki-labs.com/toki go mod tidy
|
||||
(출력 없음)
|
||||
|
||||
$ GOPRIVATE=git.toki-labs.com/toki go list -m git.toki-labs.com/toki/proto-socket/go
|
||||
git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a
|
||||
```
|
||||
|
||||
## 재개 조건
|
||||
|
||||
- `services/core/go.mod`에 repo-external `replace`를 커밋해도 되는지 승인한다.
|
||||
- 또는 `git.toki-labs.com/toki/proto-socket/go` 원격 repository의 module path와 published version을 정합하게 고친 뒤, `go mod tidy`가 `replace` 없이 성공하는 상태를 제공한다.
|
||||
- 또는 proto-socket dependency를 vendoring/submodule/workspace 등 다른 방식으로 고정하는 새 범위를 승인한다.
|
||||
|
||||
## 다음 실행 힌트
|
||||
|
||||
- 사용자 결정 후 이 task path에서 plan 루프를 재개한다: `agent-task/m-proto-socket-infrastructure-communication-rail/01_core_endpoint`.
|
||||
|
||||
## 종료 규칙
|
||||
|
||||
- 사용자가 이 stop state를 완료/PASS로 해소하면 `USER_REVIEW.md`를 해소 상태로 갱신하고, `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
|
||||
- 새 구현이 필요하면 `plan` 스킬이 `USER_REVIEW.md`를 `user_review_N.log`로 아카이브한 뒤 새 `PLAN-*-G??.md` / `CODE_REVIEW-*-G??.md`를 작성한다.
|
||||
|
|
@ -250,10 +250,10 @@ packages:
|
|||
dependency: transitive
|
||||
description:
|
||||
name: protobuf
|
||||
sha256: "68645b24e0716782e58948f8467fd42a880f255096a821f9e7d0ec625b00c84d"
|
||||
sha256: "75ec242d22e950bdcc79ee38dd520ce4ee0bc491d7fadc4ea47694604d22bf06"
|
||||
url: "https://pub.dev"
|
||||
source: hosted
|
||||
version: "3.1.0"
|
||||
version: "6.0.0"
|
||||
sky_engine:
|
||||
dependency: transitive
|
||||
description: flutter
|
||||
|
|
|
|||
|
|
@ -9,9 +9,122 @@
|
|||
|
||||
---
|
||||
|
||||
## 1. 공통 및 인프라 API
|
||||
## 1. 내부 통신 원칙
|
||||
|
||||
### 1.1 `GET /healthz`
|
||||
### 1.1 기본 표준
|
||||
- **설명**: NomadCode 내부 런타임 간 신규 통신은 `proto-socket`을 기본 표준으로 사용한다.
|
||||
- **안정성 수준**: 후보 (Candidate)
|
||||
- **적용 대상**:
|
||||
- `apps/client` ↔ `services/core`
|
||||
- NomadCode Core ↔ 내부 control/runtime adapter
|
||||
- 향후 NomadCode가 직접 소유하는 MCP/action runtime 표면
|
||||
- **예외 대상**:
|
||||
- health/readiness처럼 외부 운영 도구와 curl smoke가 직접 확인해야 하는 endpoint
|
||||
- Plane/Jira/Mattermost/IOP/OpenAI-compatible Responses/A2A처럼 외부 provider 또는 외부 실행 표면이 요구하는 native protocol
|
||||
- public callback/webhook
|
||||
- 단순 정적 조회처럼 `proto-socket` 연결 비용이 과한 read-only endpoint
|
||||
- **예외 기록 규칙**: REST/HTTP를 새로 추가하거나 유지할 때는 이 문서 또는 관련 domain rule에 예외 사유를 남긴다.
|
||||
|
||||
### 1.2 proto-socket Semantic Envelope 후보
|
||||
- **설명**: 아래 JSON은 wire schema가 아니라 proto-socket message가 가져야 할 의미 필드 후보다. 실제 source schema는 proto-socket 계약이 안정화되면 별도 schema로 승격한다.
|
||||
- **Envelope Shape 후보**:
|
||||
```json
|
||||
{
|
||||
"protocol_version": "nomadcode.proto-socket.v1",
|
||||
"id": "msg-uuid",
|
||||
"correlation_id": "request-or-flow-id",
|
||||
"type": "request",
|
||||
"channel": "task",
|
||||
"action": "task.list",
|
||||
"actor": {
|
||||
"kind": "client",
|
||||
"id": "apps/client"
|
||||
},
|
||||
"auth": {
|
||||
"scheme": "basic",
|
||||
"token_ref": "runtime-secret-ref"
|
||||
},
|
||||
"payload": {},
|
||||
"error": null,
|
||||
"meta": {
|
||||
"connection_id": "conn-id",
|
||||
"timestamp": "2026-05-30T00:00:00Z"
|
||||
}
|
||||
}
|
||||
```
|
||||
- **type 후보**:
|
||||
- `request`: 요청 메시지
|
||||
- `response`: 요청 성공 응답
|
||||
- `event`: 구독 또는 서버 발행 이벤트
|
||||
- `error`: 요청 실패 또는 channel-level 오류
|
||||
- `heartbeat`: 연결 생존 확인
|
||||
- `auth`: 연결 인증 또는 인증 갱신
|
||||
- **error 후보 필드**:
|
||||
```json
|
||||
{
|
||||
"code": "task.not_found",
|
||||
"message": "task not found",
|
||||
"retryable": false,
|
||||
"details": {}
|
||||
}
|
||||
```
|
||||
|
||||
### 1.3 REST Compatibility Map 후보
|
||||
- **설명**: 현재 REST API는 proto-socket rail 구현 전까지의 호환 표면이다. task 관련 신규 내부 기능은 proto-socket channel을 우선한다.
|
||||
|
||||
| 현재 REST/HTTP 표면 | proto-socket 후보 | 유지/전환 기준 |
|
||||
|--------------------|-------------------|----------------|
|
||||
| `GET /healthz` | 없음 | Public 운영 health check로 REST 유지 |
|
||||
| `GET /readyz` | `system.ready` event 또는 request 후보 | DB readiness와 curl smoke 때문에 REST 유지, 내부 UI에는 proto-socket 상태 이벤트를 추가 가능 |
|
||||
| `POST /api/tasks` | `task.create` request | client-core 내부 호출은 proto-socket으로 전환, REST는 smoke/compat 유지 |
|
||||
| `GET /api/tasks` | `task.list` request | client-core 내부 조회는 proto-socket으로 전환, REST는 smoke/compat 유지 |
|
||||
| `GET /api/tasks/{id}` | `task.get` request | client-core 내부 조회는 proto-socket으로 전환, REST는 smoke/compat 유지 |
|
||||
| `POST /api/tasks/{id}/enqueue` | `task.enqueue` request + `task.status.changed` event | client-core 실행 요청과 상태 반영은 proto-socket으로 전환, REST는 smoke/compat 유지 |
|
||||
| `POST /api/integrations/plane/tasks` | `workitem.plane.import` 후보 | Plane API 자체는 외부 provider REST이며, 내부 호출 표면은 proto-socket으로 추가 가능 |
|
||||
|
||||
### 1.4 Core Endpoint 후보
|
||||
- **기본 경로**: `/proto-socket`
|
||||
- **인증**: 기존 Core Basic Auth middleware 적용
|
||||
- **Wire message**: 초기 구현은 `google.protobuf.Struct` semantic envelope를 사용한다.
|
||||
- **Heartbeat**: proto-socket 기본 heartbeat를 사용하며 Core 설정 기본값은 interval 30s, wait 10s다.
|
||||
|
||||
### 1.5 Task Channel Actions (구현됨)
|
||||
- **설명**: `services/core/internal/protosocket`가 REST task API와 같은 의미를 `task` channel의 proto-socket action으로 제공한다. REST는 smoke/compat로 유지한다.
|
||||
- **안정성 수준**: 후보 (Candidate)
|
||||
- **채널**: `task`
|
||||
- **요청/응답 action 표**:
|
||||
|
||||
| action | request payload | response type | response payload |
|
||||
|--------|-----------------|---------------|------------------|
|
||||
| `task.create` | `POST /api/tasks` JSON shape (`title`, `source`, `payload`, `metadata?`, `external?`) | `response` | `{ "id", "status", "external_provider", "external_id", "task": { …full task } }` |
|
||||
| `task.list` | `{ "limit": 20 }` (생략 시 0 전달, 기본 20/최대 100 bounding은 core workflow service가 담당) | `response` | `{ "tasks": [ { …full task } ] }` |
|
||||
| `task.get` | `{ "id": "task-id" }` | `response` | `{ "task": { …full task } }` |
|
||||
| `task.enqueue` | `{ "id": "task-id" }` | `response` | `{ "id", "status": "queued", "task": { …full task } }` |
|
||||
|
||||
- **task status event (서버 발행)**:
|
||||
|
||||
| action | type | channel | payload |
|
||||
|--------|------|---------|---------|
|
||||
| `task.status.changed` | `event` | `task` | `{ "id", "type": "task.running\|task.completed\|task.failed\|task.canceled", "status", "title", "message" }` |
|
||||
|
||||
- scheduler가 내는 모든 task lifecycle event(running/completed/failed/canceled)가 notification fanout을 통해 broadcast된다. completed event의 Mattermost 알림 동작은 그대로 유지된다.
|
||||
|
||||
- **error envelope code 표** (`type: "error"`):
|
||||
|
||||
| code | 의미 | retryable | REST 대응 status |
|
||||
|------|------|-----------|------------------|
|
||||
| `task.invalid_input` | 잘못된 task 입력(`ErrInvalidTaskInput`) | false | 400 |
|
||||
| `task.conflict` | 현재 상태에서 enqueue 불가(`ErrTaskCannotBeEnqueued`) | false | 409 |
|
||||
| `task.not_found` | task 없음(`pgx.ErrNoRows`) | false | 404 |
|
||||
| `task.invalid_payload` | envelope payload 디코드 실패 | false | (REST 없음) |
|
||||
| `internal.error` | 그 외 내부 오류 | true | 500 |
|
||||
| `UNSUPPORTED_ACTION` | dispatcher에 등록되지 않은 action | false | (REST 없음) |
|
||||
|
||||
---
|
||||
|
||||
## 2. 공통 및 인프라 API
|
||||
|
||||
### 2.1 `GET /healthz`
|
||||
- **설명**: 서비스 헬스 체크.
|
||||
- **인증**: 없음 (Public)
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -22,7 +135,7 @@
|
|||
}
|
||||
```
|
||||
|
||||
### 1.2 `GET /readyz`
|
||||
### 2.2 `GET /readyz`
|
||||
- **설명**: 서비스 및 데이터베이스 준비 상태 체크.
|
||||
- **인증**: Basic Auth
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -41,9 +154,9 @@
|
|||
|
||||
---
|
||||
|
||||
## 2. 태스크 관리 API (Tasks API)
|
||||
## 3. 태스크 관리 API (Tasks API)
|
||||
|
||||
### 2.1 `POST /api/tasks`
|
||||
### 3.1 `POST /api/tasks`
|
||||
- **설명**: 새로운 태스크 생성.
|
||||
- **인증**: Basic Auth
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -72,7 +185,7 @@
|
|||
}
|
||||
```
|
||||
|
||||
### 2.2 `GET /api/tasks`
|
||||
### 3.2 `GET /api/tasks`
|
||||
- **설명**: 태스크 목록 조회.
|
||||
- **인증**: Basic Auth
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -100,7 +213,7 @@
|
|||
]
|
||||
```
|
||||
|
||||
### 2.3 `GET /api/tasks/{id}`
|
||||
### 3.3 `GET /api/tasks/{id}`
|
||||
- **설명**: 단일 태스크 세부 정보 조회.
|
||||
- **인증**: Basic Auth
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -124,7 +237,7 @@
|
|||
}
|
||||
```
|
||||
|
||||
### 2.4 `POST /api/tasks/{id}/enqueue`
|
||||
### 3.4 `POST /api/tasks/{id}/enqueue`
|
||||
- **설명**: 태스크를 큐에 진입시켜 실행 대기 상태로 변경.
|
||||
- **인증**: Basic Auth
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -138,9 +251,9 @@
|
|||
|
||||
---
|
||||
|
||||
## 3. 외부 연동 API (Integrations API)
|
||||
## 4. 외부 연동 API (Integrations API)
|
||||
|
||||
### 3.1 `POST /api/integrations/plane/tasks`
|
||||
### 4.1 `POST /api/integrations/plane/tasks`
|
||||
- **설명**: Plane 워크아이템 정보를 기반으로 새로운 태스크 생성.
|
||||
- **인증**: Basic Auth
|
||||
- **안정성 수준**: 안정 (Stable)
|
||||
|
|
@ -167,12 +280,12 @@
|
|||
|
||||
---
|
||||
|
||||
## 4. Workspace / Project Metadata 계약 후보
|
||||
## 5. Workspace / Project Metadata 계약 후보
|
||||
|
||||
> [!IMPORTANT]
|
||||
> **Workspace/Project Metadata 상태**: 현재 Go core 백엔드에 이와 관련한 별도의 Source-of-Truth 저장소나 전용 테이블 구조가 존재하지 않으며, API 명세 또한 완전히 확정되지 않았습니다. 아래는 Flutter 앱과 코어 간 향후 연동을 대비한 **계약 후보 필드 목록**입니다.
|
||||
|
||||
### 4.1 Workspace Metadata 후보 스펙
|
||||
### 5.1 Workspace Metadata 후보 스펙
|
||||
- **설명**: 에이전트 작업 공간(Workspace)에 대한 설정 및 상태 메타데이터.
|
||||
- **후보 필드**:
|
||||
- `workspace_id`: 작업 공간 고유 식별자 (UUID)
|
||||
|
|
@ -184,7 +297,7 @@
|
|||
- `status`: 현재 상태 (예: `active`, `archived`, `suspended`)
|
||||
- `settings`: 작업 공간에 특화된 동적 설정 JSON 오브젝트 (예: 자동 스케줄링 옵션, 알림 채널 정보 등)
|
||||
|
||||
### 4.2 Project Metadata 후보 스펙
|
||||
### 5.2 Project Metadata 후보 스펙
|
||||
- **설명**: 작업 공간 내 세부 프로젝트(Project) 정보.
|
||||
- **후보 필드**:
|
||||
- `project_id`: 프로젝트 고유 식별자 (UUID)
|
||||
|
|
@ -197,13 +310,13 @@
|
|||
|
||||
---
|
||||
|
||||
## 5. 클라이언트 Integration 설정 계약 후보
|
||||
## 6. 클라이언트 Integration 설정 계약 후보
|
||||
|
||||
> [!IMPORTANT]
|
||||
> 이 절은 `apps/client`의 integration boundary에서 host/plugin/transport 사이에 교환되는 설정값을 후보로 모은 것이다. 실제 source schema는 소비하는 core/client 경계가 안정화된 뒤로 미룬다.
|
||||
> 관련 코드 경계: `apps/client/lib/src/integrations/proto_socket/` 와 `apps/client/lib/src/integrations/mattermost/`.
|
||||
|
||||
### 5.1 proto-socket Endpoint 설정 후보
|
||||
### 6.1 proto-socket Endpoint 설정 후보
|
||||
- **설명**: `apps/client`가 proto-socket transport에 의존할 때 사용하는 연결/하트비트 설정. 현재 구현은 `proto_socket` Dart 패키지(local path)를 끌어 쓰며 bootstrap에서 자동 연결하지 않는다.
|
||||
- **후보 필드**:
|
||||
- `host`: 대상 서버 hostname
|
||||
|
|
@ -214,7 +327,7 @@
|
|||
- `heartbeat_wait_seconds`: heartbeat 응답 대기 한계 (기본 60)
|
||||
- `enabled`: bootstrap 시 자동 연결 활성화 여부 (기본 false)
|
||||
|
||||
### 5.2 Mattermost Push Host 책임 경계 후보
|
||||
### 6.2 Mattermost Push Host 책임 경계 후보
|
||||
- **설명**: Mattermost push 통합에서 `apps/client` host가 plugin adapter에게 위임/공급해야 하는 데이터와 콜백. plugin adapter는 platform 측 plugin singleton에 격리되어 있고, 그 외 코드는 모두 `MattermostPushClient` 인터페이스에만 의존한다.
|
||||
- **호스트 소유 필드/콜백**:
|
||||
- `server_url`: Mattermost 서버 base URL (auth 핸드오프, signing key 저장에 함께 사용)
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"github.com/nomadcode/nomadcode-core/internal/db"
|
||||
apphttp "github.com/nomadcode/nomadcode-core/internal/http"
|
||||
"github.com/nomadcode/nomadcode-core/internal/notification"
|
||||
"github.com/nomadcode/nomadcode-core/internal/protosocket"
|
||||
"github.com/nomadcode/nomadcode-core/internal/scheduler"
|
||||
"github.com/nomadcode/nomadcode-core/internal/storage"
|
||||
"github.com/nomadcode/nomadcode-core/internal/workflow"
|
||||
|
|
@ -53,7 +54,12 @@ func run(logger *slog.Logger) error {
|
|||
BaseURL: cfg.PlaneBaseURL,
|
||||
Token: cfg.PlaneToken,
|
||||
}, logger)
|
||||
notificationService := notification.NewService(mattermostClient, logger)
|
||||
protoSocketServer := protosocket.NewServer(protosocket.Config{
|
||||
HeartbeatIntervalSec: cfg.ProtoSocketHeartbeatIntervalSec,
|
||||
HeartbeatWaitSec: cfg.ProtoSocketHeartbeatWaitSec,
|
||||
}, logger)
|
||||
|
||||
notificationService := notification.NewService(mattermostClient, logger, protosocket.NewTaskEventBroadcaster(protoSocketServer))
|
||||
modelClient := modelopenai.NewClient(modelopenai.Config{
|
||||
BaseURL: cfg.ModelBaseURL,
|
||||
APIKey: cfg.ModelAPIKey,
|
||||
|
|
@ -87,9 +93,12 @@ func run(logger *slog.Logger) error {
|
|||
|
||||
workflowService := workflow.NewService(store, taskScheduler, logger)
|
||||
handler := apphttp.NewHandler(pool, workflowService, planeClient, logger)
|
||||
|
||||
protosocket.NewTaskChannels(workflowService).Register(protoSocketServer.Dispatcher())
|
||||
|
||||
server := &http.Server{
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: apphttp.NewRouter(handler, logger, apphttp.AuthConfig{Username: cfg.AuthUsername, Password: cfg.AuthPassword}),
|
||||
Handler: apphttp.NewRouter(handler, logger, apphttp.AuthConfig{Username: cfg.AuthUsername, Password: cfg.AuthPassword}, protoSocketServer, cfg.ProtoSocketPath),
|
||||
ReadHeaderTimeout: 5 * time.Second,
|
||||
}
|
||||
|
||||
|
|
@ -117,6 +126,9 @@ func run(logger *slog.Logger) error {
|
|||
if err := server.Shutdown(shutdownCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := protoSocketServer.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := taskScheduler.Stop(shutdownCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,10 +3,13 @@ module github.com/nomadcode/nomadcode-core
|
|||
go 1.25.0
|
||||
|
||||
require (
|
||||
git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a
|
||||
github.com/go-chi/chi/v5 v5.2.5
|
||||
github.com/jackc/pgx/v5 v5.9.2
|
||||
github.com/riverqueue/river v0.37.1
|
||||
github.com/riverqueue/river/riverdriver/riverpgxv5 v0.37.1
|
||||
google.golang.org/protobuf v1.36.11
|
||||
nhooyr.io/websocket v1.8.17
|
||||
)
|
||||
|
||||
require (
|
||||
|
|
|
|||
|
|
@ -1,8 +1,12 @@
|
|||
git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a h1:o9SRJC3AYzBuO+/WVsXJnW+OQFj0sySoKhWRI2bnMYM=
|
||||
git.toki-labs.com/toki/proto-socket/go v0.0.0-20260530100557-fed0da37fa1a/go.mod h1:ZQOfKXqiRUwRVWwe3tVVFSOKka31Wya8vzUfYChDo60=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug=
|
||||
github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0=
|
||||
github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
|
|
@ -55,9 +59,13 @@ golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
|
|||
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||
golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
|
||||
golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
|
||||
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
|
||||
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
|
||||
nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
|
||||
|
|
|
|||
|
|
@ -6,52 +6,58 @@ import (
|
|||
)
|
||||
|
||||
type Config struct {
|
||||
AppEnv string
|
||||
HTTPAddr string
|
||||
DatabaseURL string
|
||||
RedisURL string
|
||||
RedisKeyPrefix string
|
||||
AuthUsername string
|
||||
AuthPassword string
|
||||
ModelBaseURL string
|
||||
ModelAPIKey string
|
||||
ModelName string
|
||||
ModelContextSize int
|
||||
ModelTimeoutSec int
|
||||
A2AEdgeURL string
|
||||
A2AAgentURL string
|
||||
A2AToken string
|
||||
A2ATimeoutSec int
|
||||
MattermostBaseURL string
|
||||
MattermostToken string
|
||||
PlaneBaseURL string
|
||||
PlaneToken string
|
||||
WorkflowTaskTimeoutSec int
|
||||
AppEnv string
|
||||
HTTPAddr string
|
||||
DatabaseURL string
|
||||
RedisURL string
|
||||
RedisKeyPrefix string
|
||||
AuthUsername string
|
||||
AuthPassword string
|
||||
ModelBaseURL string
|
||||
ModelAPIKey string
|
||||
ModelName string
|
||||
ModelContextSize int
|
||||
ModelTimeoutSec int
|
||||
A2AEdgeURL string
|
||||
A2AAgentURL string
|
||||
A2AToken string
|
||||
A2ATimeoutSec int
|
||||
MattermostBaseURL string
|
||||
MattermostToken string
|
||||
PlaneBaseURL string
|
||||
PlaneToken string
|
||||
WorkflowTaskTimeoutSec int
|
||||
ProtoSocketPath string
|
||||
ProtoSocketHeartbeatIntervalSec int
|
||||
ProtoSocketHeartbeatWaitSec int
|
||||
}
|
||||
|
||||
func Load() Config {
|
||||
return Config{
|
||||
AppEnv: getEnv("APP_ENV", "local"),
|
||||
HTTPAddr: getEnv("HTTP_ADDR", ":8080"),
|
||||
DatabaseURL: os.Getenv("DATABASE_URL"),
|
||||
RedisURL: os.Getenv("REDIS_URL"),
|
||||
RedisKeyPrefix: os.Getenv("REDIS_KEY_PREFIX"),
|
||||
AuthUsername: getEnv("AUTH_USERNAME", "nomadcode"),
|
||||
AuthPassword: os.Getenv("AUTH_PASSWORD"),
|
||||
ModelBaseURL: getEnv("MODEL_BASE_URL", "http://192.168.0.91:11434"),
|
||||
ModelAPIKey: getEnv("MODEL_API_KEY", "ollama"),
|
||||
ModelName: getEnv("MODEL_NAME", "qwen3.6:35b-a3b-bf16"),
|
||||
ModelContextSize: getEnvInt("MODEL_CONTEXT_SIZE", 262144),
|
||||
ModelTimeoutSec: getEnvInt("MODEL_TIMEOUT_SEC", 300),
|
||||
A2AEdgeURL: firstEnv("A2A_EDGE_URL", "A2A_AGENT_URL"),
|
||||
A2AAgentURL: firstEnv("A2A_AGENT_URL", "A2A_EDGE_URL"),
|
||||
A2AToken: os.Getenv("A2A_TOKEN"),
|
||||
A2ATimeoutSec: getEnvInt("A2A_TIMEOUT_SEC", 300),
|
||||
MattermostBaseURL: os.Getenv("MATTERMOST_BASE_URL"),
|
||||
MattermostToken: os.Getenv("MATTERMOST_TOKEN"),
|
||||
PlaneBaseURL: os.Getenv("PLANE_BASE_URL"),
|
||||
PlaneToken: os.Getenv("PLANE_TOKEN"),
|
||||
WorkflowTaskTimeoutSec: getEnvInt("WORKFLOW_TASK_TIMEOUT_SEC", 300),
|
||||
AppEnv: getEnv("APP_ENV", "local"),
|
||||
HTTPAddr: getEnv("HTTP_ADDR", ":8080"),
|
||||
DatabaseURL: os.Getenv("DATABASE_URL"),
|
||||
RedisURL: os.Getenv("REDIS_URL"),
|
||||
RedisKeyPrefix: os.Getenv("REDIS_KEY_PREFIX"),
|
||||
AuthUsername: getEnv("AUTH_USERNAME", "nomadcode"),
|
||||
AuthPassword: os.Getenv("AUTH_PASSWORD"),
|
||||
ModelBaseURL: getEnv("MODEL_BASE_URL", "http://192.168.0.91:11434"),
|
||||
ModelAPIKey: getEnv("MODEL_API_KEY", "ollama"),
|
||||
ModelName: getEnv("MODEL_NAME", "qwen3.6:35b-a3b-bf16"),
|
||||
ModelContextSize: getEnvInt("MODEL_CONTEXT_SIZE", 262144),
|
||||
ModelTimeoutSec: getEnvInt("MODEL_TIMEOUT_SEC", 300),
|
||||
A2AEdgeURL: firstEnv("A2A_EDGE_URL", "A2A_AGENT_URL"),
|
||||
A2AAgentURL: firstEnv("A2A_AGENT_URL", "A2A_EDGE_URL"),
|
||||
A2AToken: os.Getenv("A2A_TOKEN"),
|
||||
A2ATimeoutSec: getEnvInt("A2A_TIMEOUT_SEC", 300),
|
||||
MattermostBaseURL: os.Getenv("MATTERMOST_BASE_URL"),
|
||||
MattermostToken: os.Getenv("MATTERMOST_TOKEN"),
|
||||
PlaneBaseURL: os.Getenv("PLANE_BASE_URL"),
|
||||
PlaneToken: os.Getenv("PLANE_TOKEN"),
|
||||
WorkflowTaskTimeoutSec: getEnvInt("WORKFLOW_TASK_TIMEOUT_SEC", 300),
|
||||
ProtoSocketPath: getEnv("PROTO_SOCKET_PATH", "/proto-socket"),
|
||||
ProtoSocketHeartbeatIntervalSec: getEnvInt("PROTO_SOCKET_HEARTBEAT_INTERVAL_SEC", 30),
|
||||
ProtoSocketHeartbeatWaitSec: getEnvInt("PROTO_SOCKET_HEARTBEAT_WAIT_SEC", 10),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -64,3 +64,31 @@ func TestConfigLoadsWorkflowTaskTimeout(t *testing.T) {
|
|||
t.Fatalf("WorkflowTaskTimeoutSec override: got %d, want 120", cfgOverride.WorkflowTaskTimeoutSec)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigLoadsProtoSocketDefaultsAndOverrides(t *testing.T) {
|
||||
cfg := Load()
|
||||
if cfg.ProtoSocketPath != "/proto-socket" {
|
||||
t.Fatalf("ProtoSocketPath default: got %q, want %q", cfg.ProtoSocketPath, "/proto-socket")
|
||||
}
|
||||
if cfg.ProtoSocketHeartbeatIntervalSec != 30 {
|
||||
t.Fatalf("ProtoSocketHeartbeatIntervalSec default: got %d, want 30", cfg.ProtoSocketHeartbeatIntervalSec)
|
||||
}
|
||||
if cfg.ProtoSocketHeartbeatWaitSec != 10 {
|
||||
t.Fatalf("ProtoSocketHeartbeatWaitSec default: got %d, want 10", cfg.ProtoSocketHeartbeatWaitSec)
|
||||
}
|
||||
|
||||
t.Setenv("PROTO_SOCKET_PATH", "/custom-path")
|
||||
t.Setenv("PROTO_SOCKET_HEARTBEAT_INTERVAL_SEC", "15")
|
||||
t.Setenv("PROTO_SOCKET_HEARTBEAT_WAIT_SEC", "5")
|
||||
|
||||
cfgOverride := Load()
|
||||
if cfgOverride.ProtoSocketPath != "/custom-path" {
|
||||
t.Fatalf("ProtoSocketPath override: got %q, want %q", cfgOverride.ProtoSocketPath, "/custom-path")
|
||||
}
|
||||
if cfgOverride.ProtoSocketHeartbeatIntervalSec != 15 {
|
||||
t.Fatalf("ProtoSocketHeartbeatIntervalSec override: got %d, want 15", cfgOverride.ProtoSocketHeartbeatIntervalSec)
|
||||
}
|
||||
if cfgOverride.ProtoSocketHeartbeatWaitSec != 5 {
|
||||
t.Fatalf("ProtoSocketHeartbeatWaitSec override: got %d, want 5", cfgOverride.ProtoSocketHeartbeatWaitSec)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
package http
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"crypto/subtle"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net"
|
||||
stdhttp "net/http"
|
||||
"time"
|
||||
)
|
||||
|
|
@ -17,6 +20,16 @@ func (r *statusRecorder) WriteHeader(status int) {
|
|||
r.ResponseWriter.WriteHeader(status)
|
||||
}
|
||||
|
||||
// Hijack lets WebSocket upgrades (e.g. the proto-socket route) take over the
|
||||
// connection even though the response writer is wrapped for status logging.
|
||||
func (r *statusRecorder) Hijack() (net.Conn, *bufio.ReadWriter, error) {
|
||||
hijacker, ok := r.ResponseWriter.(stdhttp.Hijacker)
|
||||
if !ok {
|
||||
return nil, nil, errors.New("underlying ResponseWriter does not implement http.Hijacker")
|
||||
}
|
||||
return hijacker.Hijack()
|
||||
}
|
||||
|
||||
func loggingMiddleware(logger *slog.Logger) func(stdhttp.Handler) stdhttp.Handler {
|
||||
return func(next stdhttp.Handler) stdhttp.Handler {
|
||||
return stdhttp.HandlerFunc(func(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
||||
|
|
|
|||
|
|
@ -44,10 +44,14 @@ func TestBasicAuthMiddlewareAcceptsValidCredentials(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestRouterKeepsHealthzPublicAndProtectsAPI(t *testing.T) {
|
||||
protoSocketMock := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
|
||||
router := NewRouter(NewHandler(nil, nil, nil, nil), nil, AuthConfig{
|
||||
Username: "nomadcode",
|
||||
Password: "secret",
|
||||
})
|
||||
}, protoSocketMock, "/proto-socket")
|
||||
|
||||
healthReq := httptest.NewRequest(http.MethodGet, "/healthz", nil)
|
||||
healthRec := httptest.NewRecorder()
|
||||
|
|
@ -64,6 +68,25 @@ func TestRouterKeepsHealthzPublicAndProtectsAPI(t *testing.T) {
|
|||
if apiRec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("expected /api/tasks status %d, got %d", http.StatusUnauthorized, apiRec.Code)
|
||||
}
|
||||
|
||||
// Verify /proto-socket is protected
|
||||
wsReq := httptest.NewRequest(http.MethodGet, "/proto-socket", nil)
|
||||
wsRec := httptest.NewRecorder()
|
||||
router.ServeHTTP(wsRec, wsReq)
|
||||
|
||||
if wsRec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("expected /proto-socket status %d, got %d", http.StatusUnauthorized, wsRec.Code)
|
||||
}
|
||||
|
||||
// Verify /proto-socket works with valid credentials
|
||||
wsAuthReq := httptest.NewRequest(http.MethodGet, "/proto-socket", nil)
|
||||
wsAuthReq.SetBasicAuth("nomadcode", "secret")
|
||||
wsAuthRec := httptest.NewRecorder()
|
||||
router.ServeHTTP(wsAuthRec, wsAuthReq)
|
||||
|
||||
if wsAuthRec.Code != http.StatusOK {
|
||||
t.Fatalf("expected /proto-socket with auth status %d, got %d", http.StatusOK, wsAuthRec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func okHandler() http.Handler {
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ type AuthConfig struct {
|
|||
Password string
|
||||
}
|
||||
|
||||
func NewRouter(handler *Handler, logger *slog.Logger, auth AuthConfig) stdhttp.Handler {
|
||||
func NewRouter(handler *Handler, logger *slog.Logger, auth AuthConfig, protoSocket stdhttp.Handler, protoSocketPath string) stdhttp.Handler {
|
||||
r := chi.NewRouter()
|
||||
r.Use(chimiddleware.RequestID)
|
||||
r.Use(chimiddleware.RealIP)
|
||||
|
|
@ -26,6 +26,10 @@ func NewRouter(handler *Handler, logger *slog.Logger, auth AuthConfig) stdhttp.H
|
|||
protected.Use(basicAuthMiddleware(auth))
|
||||
protected.Get("/readyz", handler.Readyz)
|
||||
|
||||
if protoSocket != nil && protoSocketPath != "" {
|
||||
protected.Handle(protoSocketPath, protoSocket)
|
||||
}
|
||||
|
||||
protected.Route("/api", func(r chi.Router) {
|
||||
r.Post("/integrations/plane/tasks", handler.CreatePlaneTask)
|
||||
r.Post("/tasks", handler.CreateTask)
|
||||
|
|
|
|||
|
|
@ -1,5 +1,13 @@
|
|||
package notification
|
||||
|
||||
import "context"
|
||||
|
||||
// TaskEventSink receives every task event so non-Mattermost consumers (e.g. the
|
||||
// proto-socket broadcaster) can fan out task status changes.
|
||||
type TaskEventSink interface {
|
||||
HandleTaskEvent(context.Context, TaskEvent) error
|
||||
}
|
||||
|
||||
type TaskNotification struct {
|
||||
TaskID string
|
||||
Title string
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package notification
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
|
||||
"github.com/nomadcode/nomadcode-core/internal/adapters/mattermost"
|
||||
|
|
@ -10,12 +11,14 @@ import (
|
|||
type Service struct {
|
||||
mattermost *mattermost.Client
|
||||
logger *slog.Logger
|
||||
sinks []TaskEventSink
|
||||
}
|
||||
|
||||
func NewService(mattermostClient *mattermost.Client, logger *slog.Logger) *Service {
|
||||
func NewService(mattermostClient *mattermost.Client, logger *slog.Logger, sinks ...TaskEventSink) *Service {
|
||||
return &Service{
|
||||
mattermost: mattermostClient,
|
||||
logger: logger,
|
||||
sinks: append([]TaskEventSink{}, sinks...),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -34,18 +37,21 @@ func (s *Service) NotifyTaskEvent(ctx context.Context, input TaskEvent) error {
|
|||
s.logger.Info("task event notification requested", "type", input.Type, "task_id", input.TaskID)
|
||||
}
|
||||
|
||||
if input.Type != TaskEventCompleted {
|
||||
return nil
|
||||
var errs []error
|
||||
for _, sink := range s.sinks {
|
||||
if err := sink.HandleTaskEvent(ctx, input); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
|
||||
if s.mattermost == nil {
|
||||
return nil
|
||||
if input.Type == TaskEventCompleted && s.mattermost != nil {
|
||||
errs = append(errs, s.mattermost.SendTaskNotification(ctx, mattermost.TaskNotificationInput{
|
||||
TaskID: input.TaskID,
|
||||
Title: input.Title,
|
||||
Status: input.Status,
|
||||
Message: input.Message,
|
||||
}))
|
||||
}
|
||||
|
||||
return s.mattermost.SendTaskNotification(ctx, mattermost.TaskNotificationInput{
|
||||
TaskID: input.TaskID,
|
||||
Title: input.Title,
|
||||
Status: input.Status,
|
||||
Message: input.Message,
|
||||
})
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,12 +2,23 @@ package notification_test
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"testing"
|
||||
|
||||
"github.com/nomadcode/nomadcode-core/internal/notification"
|
||||
)
|
||||
|
||||
type recordingSink struct {
|
||||
events []notification.TaskEvent
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *recordingSink) HandleTaskEvent(_ context.Context, event notification.TaskEvent) error {
|
||||
s.events = append(s.events, event)
|
||||
return s.err
|
||||
}
|
||||
|
||||
type logRecord struct {
|
||||
msg string
|
||||
args map[string]any
|
||||
|
|
@ -79,6 +90,52 @@ func TestNotifyTaskCompletedUsesTaskEventCompleted(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestNotifyTaskEventFansOutToSinksForRunningAndFailed(t *testing.T) {
|
||||
logger := slog.New(&spyHandler{})
|
||||
sink := &recordingSink{}
|
||||
service := notification.NewService(nil, logger, sink)
|
||||
|
||||
events := []notification.TaskEvent{
|
||||
{Type: notification.TaskEventRunning, TaskID: "task-1", Status: "running"},
|
||||
{Type: notification.TaskEventFailed, TaskID: "task-2", Status: "failed"},
|
||||
{Type: notification.TaskEventCompleted, TaskID: "task-3", Status: "completed"},
|
||||
}
|
||||
for _, e := range events {
|
||||
if err := service.NotifyTaskEvent(context.Background(), e); err != nil {
|
||||
t.Fatalf("unexpected error for %s: %v", e.Type, err)
|
||||
}
|
||||
}
|
||||
|
||||
if len(sink.events) != 3 {
|
||||
t.Fatalf("expected sink to receive 3 events, got %d", len(sink.events))
|
||||
}
|
||||
if sink.events[0].Type != notification.TaskEventRunning || sink.events[1].Type != notification.TaskEventFailed {
|
||||
t.Errorf("sink did not receive non-completed events: %+v", sink.events)
|
||||
}
|
||||
if sink.events[2].Type != notification.TaskEventCompleted {
|
||||
t.Errorf("sink did not receive completed event: %+v", sink.events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifyTaskEventReturnsSinkErrors(t *testing.T) {
|
||||
logger := slog.New(&spyHandler{})
|
||||
sinkErr := errors.New("sink boom")
|
||||
sink := &recordingSink{err: sinkErr}
|
||||
service := notification.NewService(nil, logger, sink)
|
||||
|
||||
err := service.NotifyTaskEvent(context.Background(), notification.TaskEvent{
|
||||
Type: notification.TaskEventRunning,
|
||||
TaskID: "task-1",
|
||||
Status: "running",
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("expected sink error to be returned")
|
||||
}
|
||||
if !errors.Is(err, sinkErr) {
|
||||
t.Errorf("expected joined error to wrap sink error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifyTaskEventAllowsFailedWithoutMattermost(t *testing.T) {
|
||||
spy := &spyHandler{}
|
||||
logger := slog.New(spy)
|
||||
|
|
|
|||
59
services/core/internal/protosocket/dispatcher.go
Normal file
59
services/core/internal/protosocket/dispatcher.go
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type HandlerFunc func(context.Context, Envelope) Envelope
|
||||
|
||||
type Dispatcher struct {
|
||||
mu sync.RWMutex
|
||||
handlers map[string]HandlerFunc
|
||||
}
|
||||
|
||||
func NewDispatcher() *Dispatcher {
|
||||
return &Dispatcher{
|
||||
handlers: make(map[string]HandlerFunc),
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Dispatcher) Register(action string, handler HandlerFunc) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
d.handlers[action] = handler
|
||||
}
|
||||
|
||||
func (d *Dispatcher) Dispatch(ctx context.Context, env Envelope) Envelope {
|
||||
d.mu.RLock()
|
||||
handler, ok := d.handlers[env.Action]
|
||||
d.mu.RUnlock()
|
||||
|
||||
if !ok {
|
||||
return Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: generateID(),
|
||||
CorrelationID: env.ID,
|
||||
Type: "error",
|
||||
Channel: env.Channel,
|
||||
Action: env.Action,
|
||||
Error: &EnvelopeError{
|
||||
Code: "UNSUPPORTED_ACTION",
|
||||
Message: fmt.Sprintf("unsupported action: %s", env.Action),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
return handler(ctx, env)
|
||||
}
|
||||
|
||||
func generateID() string {
|
||||
b := make([]byte, 16)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
return "error-id-fallback"
|
||||
}
|
||||
return hex.EncodeToString(b)
|
||||
}
|
||||
121
services/core/internal/protosocket/envelope.go
Normal file
121
services/core/internal/protosocket/envelope.go
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"google.golang.org/protobuf/types/known/structpb"
|
||||
)
|
||||
|
||||
const ProtocolVersion = "nomadcode.proto-socket.v1"
|
||||
|
||||
type Envelope struct {
|
||||
ProtocolVersion string `json:"protocol_version"`
|
||||
ID string `json:"id"`
|
||||
CorrelationID string `json:"correlation_id"`
|
||||
Type string `json:"type"`
|
||||
Channel string `json:"channel"`
|
||||
Action string `json:"action"`
|
||||
Actor map[string]any `json:"actor,omitempty"`
|
||||
Auth map[string]any `json:"auth,omitempty"`
|
||||
Payload map[string]any `json:"payload,omitempty"`
|
||||
Error *EnvelopeError `json:"error,omitempty"`
|
||||
Meta map[string]any `json:"meta,omitempty"`
|
||||
}
|
||||
|
||||
type EnvelopeError struct {
|
||||
Code string `json:"code"`
|
||||
Message string `json:"message"`
|
||||
Retryable bool `json:"retryable"`
|
||||
Details map[string]any `json:"details,omitempty"`
|
||||
}
|
||||
|
||||
func (e Envelope) ToStruct() (*structpb.Struct, error) {
|
||||
m := map[string]any{
|
||||
"protocol_version": e.ProtocolVersion,
|
||||
"id": e.ID,
|
||||
"correlation_id": e.CorrelationID,
|
||||
"type": e.Type,
|
||||
"channel": e.Channel,
|
||||
"action": e.Action,
|
||||
}
|
||||
if e.Actor != nil {
|
||||
m["actor"] = e.Actor
|
||||
}
|
||||
if e.Auth != nil {
|
||||
m["auth"] = e.Auth
|
||||
}
|
||||
if e.Payload != nil {
|
||||
m["payload"] = e.Payload
|
||||
}
|
||||
if e.Meta != nil {
|
||||
m["meta"] = e.Meta
|
||||
}
|
||||
if e.Error != nil {
|
||||
errMap := map[string]any{
|
||||
"code": e.Error.Code,
|
||||
"message": e.Error.Message,
|
||||
"retryable": e.Error.Retryable,
|
||||
}
|
||||
if e.Error.Details != nil {
|
||||
errMap["details"] = e.Error.Details
|
||||
}
|
||||
m["error"] = errMap
|
||||
}
|
||||
return structpb.NewStruct(m)
|
||||
}
|
||||
|
||||
func EnvelopeFromStruct(s *structpb.Struct) (Envelope, error) {
|
||||
var env Envelope
|
||||
if s == nil {
|
||||
return env, fmt.Errorf("nil struct")
|
||||
}
|
||||
m := s.AsMap()
|
||||
|
||||
if v, ok := m["protocol_version"].(string); ok {
|
||||
env.ProtocolVersion = v
|
||||
}
|
||||
if v, ok := m["id"].(string); ok {
|
||||
env.ID = v
|
||||
}
|
||||
if v, ok := m["correlation_id"].(string); ok {
|
||||
env.CorrelationID = v
|
||||
}
|
||||
if v, ok := m["type"].(string); ok {
|
||||
env.Type = v
|
||||
}
|
||||
if v, ok := m["channel"].(string); ok {
|
||||
env.Channel = v
|
||||
}
|
||||
if v, ok := m["action"].(string); ok {
|
||||
env.Action = v
|
||||
}
|
||||
if v, ok := m["actor"].(map[string]any); ok {
|
||||
env.Actor = v
|
||||
}
|
||||
if v, ok := m["auth"].(map[string]any); ok {
|
||||
env.Auth = v
|
||||
}
|
||||
if v, ok := m["payload"].(map[string]any); ok {
|
||||
env.Payload = v
|
||||
}
|
||||
if v, ok := m["meta"].(map[string]any); ok {
|
||||
env.Meta = v
|
||||
}
|
||||
if errVal, ok := m["error"].(map[string]any); ok {
|
||||
var envErr EnvelopeError
|
||||
if code, ok := errVal["code"].(string); ok {
|
||||
envErr.Code = code
|
||||
}
|
||||
if msg, ok := errVal["message"].(string); ok {
|
||||
envErr.Message = msg
|
||||
}
|
||||
if ret, ok := errVal["retryable"].(bool); ok {
|
||||
envErr.Retryable = ret
|
||||
}
|
||||
if details, ok := errVal["details"].(map[string]any); ok {
|
||||
envErr.Details = details
|
||||
}
|
||||
env.Error = &envErr
|
||||
}
|
||||
return env, nil
|
||||
}
|
||||
116
services/core/internal/protosocket/envelope_test.go
Normal file
116
services/core/internal/protosocket/envelope_test.go
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEnvelopeStructRoundTrip(t *testing.T) {
|
||||
original := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: "msg-123",
|
||||
CorrelationID: "corr-456",
|
||||
Type: "request",
|
||||
Channel: "tasks",
|
||||
Action: "create",
|
||||
Actor: map[string]any{"id": "user-1"},
|
||||
Auth: map[string]any{"token": "secret"},
|
||||
Payload: map[string]any{"title": "hello"},
|
||||
Meta: map[string]any{"source": "client"},
|
||||
Error: &EnvelopeError{
|
||||
Code: "SOME_ERROR",
|
||||
Message: "An error occurred",
|
||||
Retryable: true,
|
||||
Details: map[string]any{"retry": true},
|
||||
},
|
||||
}
|
||||
|
||||
s, err := original.ToStruct()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to convert Envelope to Struct: %v", err)
|
||||
}
|
||||
|
||||
roundtrip, err := EnvelopeFromStruct(s)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to convert Struct back to Envelope: %v", err)
|
||||
}
|
||||
|
||||
if roundtrip.ProtocolVersion != original.ProtocolVersion {
|
||||
t.Errorf("ProtocolVersion mismatch: got %q, want %q", roundtrip.ProtocolVersion, original.ProtocolVersion)
|
||||
}
|
||||
if roundtrip.ID != original.ID {
|
||||
t.Errorf("ID mismatch: got %q, want %q", roundtrip.ID, original.ID)
|
||||
}
|
||||
if roundtrip.CorrelationID != original.CorrelationID {
|
||||
t.Errorf("CorrelationID mismatch: got %q, want %q", roundtrip.CorrelationID, original.CorrelationID)
|
||||
}
|
||||
if roundtrip.Type != original.Type {
|
||||
t.Errorf("Type mismatch: got %q, want %q", roundtrip.Type, original.Type)
|
||||
}
|
||||
if roundtrip.Channel != original.Channel {
|
||||
t.Errorf("Channel mismatch: got %q, want %q", roundtrip.Channel, original.Channel)
|
||||
}
|
||||
if roundtrip.Action != original.Action {
|
||||
t.Errorf("Action mismatch: got %q, want %q", roundtrip.Action, original.Action)
|
||||
}
|
||||
|
||||
if roundtrip.Actor["id"] != "user-1" {
|
||||
t.Errorf("Actor mismatch: got %v", roundtrip.Actor)
|
||||
}
|
||||
if roundtrip.Auth["token"] != "secret" {
|
||||
t.Errorf("Auth mismatch: got %v", roundtrip.Auth)
|
||||
}
|
||||
if roundtrip.Payload["title"] != "hello" {
|
||||
t.Errorf("Payload mismatch: got %v", roundtrip.Payload)
|
||||
}
|
||||
if roundtrip.Meta["source"] != "client" {
|
||||
t.Errorf("Meta mismatch: got %v", roundtrip.Meta)
|
||||
}
|
||||
|
||||
if roundtrip.Error == nil {
|
||||
t.Fatal("Error should not be nil")
|
||||
}
|
||||
if roundtrip.Error.Code != "SOME_ERROR" {
|
||||
t.Errorf("Error.Code mismatch: got %q, want %q", roundtrip.Error.Code, "SOME_ERROR")
|
||||
}
|
||||
if roundtrip.Error.Message != "An error occurred" {
|
||||
t.Errorf("Error.Message mismatch: got %q, want %q", roundtrip.Error.Message, "An error occurred")
|
||||
}
|
||||
if !roundtrip.Error.Retryable {
|
||||
t.Errorf("Error.Retryable mismatch: got false, want true")
|
||||
}
|
||||
if roundtrip.Error.Details["retry"] != true {
|
||||
t.Errorf("Error.Details mismatch: got %v", roundtrip.Error.Details)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatcherReturnsUnsupportedActionErrorEnvelope(t *testing.T) {
|
||||
dispatcher := NewDispatcher()
|
||||
ctx := context.Background()
|
||||
|
||||
request := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: "req-1",
|
||||
Type: "request",
|
||||
Channel: "tasks",
|
||||
Action: "unknown-action",
|
||||
}
|
||||
|
||||
response := dispatcher.Dispatch(ctx, request)
|
||||
|
||||
if response.Type != "error" {
|
||||
t.Errorf("expected response type 'error', got %q", response.Type)
|
||||
}
|
||||
if response.CorrelationID != "req-1" {
|
||||
t.Errorf("expected CorrelationID 'req-1', got %q", response.CorrelationID)
|
||||
}
|
||||
if response.Error == nil {
|
||||
t.Fatal("expected non-nil Error in response")
|
||||
}
|
||||
if response.Error.Code != "UNSUPPORTED_ACTION" {
|
||||
t.Errorf("expected error code 'UNSUPPORTED_ACTION', got %q", response.Error.Code)
|
||||
}
|
||||
if response.Error.Retryable {
|
||||
t.Error("expected Retryable to be false for UNSUPPORTED_ACTION")
|
||||
}
|
||||
}
|
||||
75
services/core/internal/protosocket/events.go
Normal file
75
services/core/internal/protosocket/events.go
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/nomadcode/nomadcode-core/internal/notification"
|
||||
)
|
||||
|
||||
// SuccessResponse builds a response envelope correlated to a request.
|
||||
func SuccessResponse(req Envelope, payload map[string]any) Envelope {
|
||||
return Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: generateID(),
|
||||
CorrelationID: req.ID,
|
||||
Type: "response",
|
||||
Channel: req.Channel,
|
||||
Action: req.Action,
|
||||
Payload: payload,
|
||||
}
|
||||
}
|
||||
|
||||
// ErrorResponse builds an error envelope correlated to a request.
|
||||
func ErrorResponse(req Envelope, code, message string, retryable bool) Envelope {
|
||||
return Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: generateID(),
|
||||
CorrelationID: req.ID,
|
||||
Type: "error",
|
||||
Channel: req.Channel,
|
||||
Action: req.Action,
|
||||
Error: &EnvelopeError{
|
||||
Code: code,
|
||||
Message: message,
|
||||
Retryable: retryable,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Broadcaster publishes an envelope to all connected clients.
|
||||
type Broadcaster interface {
|
||||
BroadcastEnvelope(ctx context.Context, env Envelope) error
|
||||
}
|
||||
|
||||
// TaskEventBroadcaster adapts notification task events into proto-socket
|
||||
// task.status.changed events. It implements notification.TaskEventSink.
|
||||
type TaskEventBroadcaster struct {
|
||||
broadcaster Broadcaster
|
||||
}
|
||||
|
||||
func NewTaskEventBroadcaster(broadcaster Broadcaster) *TaskEventBroadcaster {
|
||||
return &TaskEventBroadcaster{broadcaster: broadcaster}
|
||||
}
|
||||
|
||||
func (b *TaskEventBroadcaster) HandleTaskEvent(ctx context.Context, event notification.TaskEvent) error {
|
||||
return b.broadcaster.BroadcastEnvelope(ctx, TaskStatusChangedEnvelope(event))
|
||||
}
|
||||
|
||||
// TaskStatusChangedEnvelope maps a notification task event onto the
|
||||
// task.status.changed event envelope.
|
||||
func TaskStatusChangedEnvelope(event notification.TaskEvent) Envelope {
|
||||
return Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: generateID(),
|
||||
Type: "event",
|
||||
Channel: "task",
|
||||
Action: "task.status.changed",
|
||||
Payload: map[string]any{
|
||||
"id": event.TaskID,
|
||||
"type": string(event.Type),
|
||||
"status": event.Status,
|
||||
"title": event.Title,
|
||||
"message": event.Message,
|
||||
},
|
||||
}
|
||||
}
|
||||
153
services/core/internal/protosocket/server.go
Normal file
153
services/core/internal/protosocket/server.go
Normal file
|
|
@ -0,0 +1,153 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sync"
|
||||
|
||||
toki "git.toki-labs.com/toki/proto-socket/go"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"google.golang.org/protobuf/types/known/structpb"
|
||||
"nhooyr.io/websocket"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
HeartbeatIntervalSec int
|
||||
HeartbeatWaitSec int
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
cfg Config
|
||||
dispatcher *Dispatcher
|
||||
logger *slog.Logger
|
||||
mu sync.Mutex
|
||||
clients map[*toki.WsClient]struct{}
|
||||
}
|
||||
|
||||
func NewServer(cfg Config, logger *slog.Logger) *Server {
|
||||
return &Server{
|
||||
cfg: cfg,
|
||||
dispatcher: NewDispatcher(),
|
||||
logger: logger,
|
||||
clients: make(map[*toki.WsClient]struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
opts := &websocket.AcceptOptions{
|
||||
InsecureSkipVerify: true,
|
||||
}
|
||||
conn, err := websocket.Accept(w, r, opts)
|
||||
if err != nil {
|
||||
s.logger.Error("failed to accept websocket connection", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
client := toki.NewWsClient(
|
||||
conn,
|
||||
s.cfg.HeartbeatIntervalSec,
|
||||
s.cfg.HeartbeatWaitSec,
|
||||
ParserMap(),
|
||||
)
|
||||
|
||||
s.mu.Lock()
|
||||
s.clients[client] = struct{}{}
|
||||
s.mu.Unlock()
|
||||
|
||||
client.AddDisconnectListener(func(c *toki.WsClient) {
|
||||
s.mu.Lock()
|
||||
delete(s.clients, c)
|
||||
s.mu.Unlock()
|
||||
s.logger.Info("websocket client disconnected")
|
||||
})
|
||||
|
||||
toki.AddRequestListenerTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, func(req *structpb.Struct) (*structpb.Struct, error) {
|
||||
env, err := EnvelopeFromStruct(req)
|
||||
if err != nil {
|
||||
errEnv := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: generateID(),
|
||||
CorrelationID: "",
|
||||
Type: "error",
|
||||
Error: &EnvelopeError{
|
||||
Code: "INVALID_ENVELOPE",
|
||||
Message: "failed to parse envelope: " + err.Error(),
|
||||
},
|
||||
}
|
||||
res, _ := errEnv.ToStruct()
|
||||
return res, nil
|
||||
}
|
||||
|
||||
resEnv := s.dispatcher.Dispatch(r.Context(), env)
|
||||
resStruct, err := resEnv.ToStruct()
|
||||
if err != nil {
|
||||
errEnv := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: generateID(),
|
||||
CorrelationID: env.ID,
|
||||
Type: "error",
|
||||
Error: &EnvelopeError{
|
||||
Code: "SERIALIZATION_ERROR",
|
||||
Message: "failed to serialize response envelope: " + err.Error(),
|
||||
},
|
||||
}
|
||||
res, _ := errEnv.ToStruct()
|
||||
return res, nil
|
||||
}
|
||||
|
||||
return resStruct, nil
|
||||
})
|
||||
|
||||
s.logger.Info("websocket client connected successfully")
|
||||
}
|
||||
|
||||
func (s *Server) Close() error {
|
||||
s.mu.Lock()
|
||||
clients := make([]*toki.WsClient, 0, len(s.clients))
|
||||
for cl := range s.clients {
|
||||
clients = append(clients, cl)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
for _, cl := range clients {
|
||||
_ = cl.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) BroadcastEnvelope(ctx context.Context, env Envelope) error {
|
||||
structMsg, err := env.ToStruct()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
clients := make([]*toki.WsClient, 0, len(s.clients))
|
||||
for cl := range s.clients {
|
||||
clients = append(clients, cl)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
for _, cl := range clients {
|
||||
if cl.IsAlive() {
|
||||
if err := cl.Send(structMsg); err != nil {
|
||||
s.logger.Error("failed to send broadcast to client", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) Dispatcher() *Dispatcher {
|
||||
return s.dispatcher
|
||||
}
|
||||
|
||||
func ParserMap() toki.ParserMap {
|
||||
return toki.ParserMap{
|
||||
toki.TypeNameOf(&structpb.Struct{}): func(b []byte) (proto.Message, error) {
|
||||
msg := &structpb.Struct{}
|
||||
return msg, proto.Unmarshal(b, msg)
|
||||
},
|
||||
}
|
||||
}
|
||||
138
services/core/internal/protosocket/server_test.go
Normal file
138
services/core/internal/protosocket/server_test.go
Normal file
|
|
@ -0,0 +1,138 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
toki "git.toki-labs.com/toki/proto-socket/go"
|
||||
"google.golang.org/protobuf/types/known/structpb"
|
||||
"nhooyr.io/websocket"
|
||||
)
|
||||
|
||||
func TestServerAcceptsWebSocketAndClosesClients(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
srv := NewServer(Config{
|
||||
HeartbeatIntervalSec: 0,
|
||||
HeartbeatWaitSec: 0,
|
||||
}, logger)
|
||||
|
||||
testServer := httptest.NewServer(srv)
|
||||
defer testServer.Close()
|
||||
|
||||
wsURL := strings.Replace(testServer.URL, "http", "ws", 1)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
conn, _, err := websocket.Dial(ctx, wsURL, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to dial websocket: %v", err)
|
||||
}
|
||||
defer conn.Close(websocket.StatusNormalClosure, "")
|
||||
|
||||
client := toki.NewWsClient(conn, 0, 0, ParserMap())
|
||||
defer client.Close()
|
||||
|
||||
if !client.IsAlive() {
|
||||
t.Error("client should be alive after connection")
|
||||
}
|
||||
|
||||
disconnected := make(chan struct{}, 1)
|
||||
client.AddDisconnectListener(func(c *toki.WsClient) {
|
||||
disconnected <- struct{}{}
|
||||
})
|
||||
|
||||
if err := srv.Close(); err != nil {
|
||||
t.Errorf("failed to close server: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-disconnected:
|
||||
// success
|
||||
case <-time.After(1 * time.Second):
|
||||
t.Error("timed out waiting for client to disconnect after server close")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerRespondsWithErrorEnvelopeForUnsupportedAction(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
srv := NewServer(Config{
|
||||
HeartbeatIntervalSec: 0,
|
||||
HeartbeatWaitSec: 0,
|
||||
}, logger)
|
||||
defer srv.Close()
|
||||
|
||||
testServer := httptest.NewServer(srv)
|
||||
defer testServer.Close()
|
||||
|
||||
wsURL := strings.Replace(testServer.URL, "http", "ws", 1)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
conn, _, err := websocket.Dial(ctx, wsURL, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to dial websocket: %v", err)
|
||||
}
|
||||
defer conn.Close(websocket.StatusNormalClosure, "")
|
||||
|
||||
client := toki.NewWsClient(conn, 0, 0, ParserMap())
|
||||
defer client.Close()
|
||||
|
||||
requestEnv := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: "req-999",
|
||||
Type: "request",
|
||||
Channel: "tasks",
|
||||
Action: "invalid-unsupported-action",
|
||||
}
|
||||
reqStruct, err := requestEnv.ToStruct()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create request struct: %v", err)
|
||||
}
|
||||
|
||||
resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 1*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to send request: %v", err)
|
||||
}
|
||||
|
||||
responseEnv, err := EnvelopeFromStruct(resStruct)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to parse response struct: %v", err)
|
||||
}
|
||||
|
||||
if responseEnv.Type != "error" {
|
||||
t.Errorf("expected response type 'error', got %q", responseEnv.Type)
|
||||
}
|
||||
if responseEnv.CorrelationID != "req-999" {
|
||||
t.Errorf("expected CorrelationID 'req-999', got %q", responseEnv.CorrelationID)
|
||||
}
|
||||
if responseEnv.Error == nil {
|
||||
t.Fatal("expected error in response env")
|
||||
}
|
||||
if responseEnv.Error.Code != "UNSUPPORTED_ACTION" {
|
||||
t.Errorf("expected error code 'UNSUPPORTED_ACTION', got %q", responseEnv.Error.Code)
|
||||
}
|
||||
if responseEnv.Error.Retryable {
|
||||
t.Error("expected Retryable to be false for UNSUPPORTED_ACTION")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerHeartbeatConfigSupport(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
|
||||
srvZero := NewServer(Config{
|
||||
HeartbeatIntervalSec: 0,
|
||||
HeartbeatWaitSec: 0,
|
||||
}, logger)
|
||||
srvZero.Close()
|
||||
|
||||
srvNonZero := NewServer(Config{
|
||||
HeartbeatIntervalSec: 10,
|
||||
HeartbeatWaitSec: 2,
|
||||
}, logger)
|
||||
srvNonZero.Close()
|
||||
}
|
||||
182
services/core/internal/protosocket/tasks.go
Normal file
182
services/core/internal/protosocket/tasks.go
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"github.com/nomadcode/nomadcode-core/internal/storage"
|
||||
"github.com/nomadcode/nomadcode-core/internal/workflow"
|
||||
)
|
||||
|
||||
// TaskService is the subset of the workflow service required to map REST task
|
||||
// semantics onto proto-socket actions.
|
||||
type TaskService interface {
|
||||
CreateTask(context.Context, workflow.CreateTaskInput) (storage.Task, error)
|
||||
ListTasks(context.Context, int32) ([]storage.Task, error)
|
||||
GetTask(context.Context, string) (storage.Task, error)
|
||||
EnqueueTask(context.Context, string) (storage.Task, error)
|
||||
}
|
||||
|
||||
// TaskChannels registers task.* request handlers on a dispatcher.
|
||||
type TaskChannels struct {
|
||||
tasks TaskService
|
||||
}
|
||||
|
||||
func NewTaskChannels(tasks TaskService) *TaskChannels {
|
||||
return &TaskChannels{tasks: tasks}
|
||||
}
|
||||
|
||||
func (c *TaskChannels) Register(dispatcher *Dispatcher) {
|
||||
dispatcher.Register("task.create", c.handleCreate)
|
||||
dispatcher.Register("task.list", c.handleList)
|
||||
dispatcher.Register("task.get", c.handleGet)
|
||||
dispatcher.Register("task.enqueue", c.handleEnqueue)
|
||||
}
|
||||
|
||||
type listRequest struct {
|
||||
Limit int32 `json:"limit"`
|
||||
}
|
||||
|
||||
type idRequest struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
func (c *TaskChannels) handleCreate(ctx context.Context, req Envelope) Envelope {
|
||||
input, err := payloadAs[workflow.CreateTaskInput](req.Payload)
|
||||
if err != nil {
|
||||
return invalidPayloadResponse(req, err)
|
||||
}
|
||||
|
||||
task, err := c.tasks.CreateTask(ctx, input)
|
||||
if err != nil {
|
||||
return mapTaskError(req, err)
|
||||
}
|
||||
|
||||
taskMap, err := taskToMap(task)
|
||||
if err != nil {
|
||||
return ErrorResponse(req, "internal.error", "internal server error", true)
|
||||
}
|
||||
|
||||
return SuccessResponse(req, map[string]any{
|
||||
"id": task.ID,
|
||||
"status": task.Status,
|
||||
"external_provider": stringOrEmpty(task.ExternalProvider),
|
||||
"external_id": stringOrEmpty(task.ExternalID),
|
||||
"task": taskMap,
|
||||
})
|
||||
}
|
||||
|
||||
func (c *TaskChannels) handleList(ctx context.Context, req Envelope) Envelope {
|
||||
input, err := payloadAs[listRequest](req.Payload)
|
||||
if err != nil {
|
||||
return invalidPayloadResponse(req, err)
|
||||
}
|
||||
|
||||
tasks, err := c.tasks.ListTasks(ctx, input.Limit)
|
||||
if err != nil {
|
||||
return mapTaskError(req, err)
|
||||
}
|
||||
|
||||
encoded := make([]any, 0, len(tasks))
|
||||
for _, task := range tasks {
|
||||
taskMap, err := taskToMap(task)
|
||||
if err != nil {
|
||||
return ErrorResponse(req, "internal.error", "internal server error", true)
|
||||
}
|
||||
encoded = append(encoded, taskMap)
|
||||
}
|
||||
|
||||
return SuccessResponse(req, map[string]any{"tasks": encoded})
|
||||
}
|
||||
|
||||
func (c *TaskChannels) handleGet(ctx context.Context, req Envelope) Envelope {
|
||||
input, err := payloadAs[idRequest](req.Payload)
|
||||
if err != nil {
|
||||
return invalidPayloadResponse(req, err)
|
||||
}
|
||||
|
||||
task, err := c.tasks.GetTask(ctx, input.ID)
|
||||
if err != nil {
|
||||
return mapTaskError(req, err)
|
||||
}
|
||||
|
||||
taskMap, err := taskToMap(task)
|
||||
if err != nil {
|
||||
return ErrorResponse(req, "internal.error", "internal server error", true)
|
||||
}
|
||||
|
||||
return SuccessResponse(req, map[string]any{"task": taskMap})
|
||||
}
|
||||
|
||||
func (c *TaskChannels) handleEnqueue(ctx context.Context, req Envelope) Envelope {
|
||||
input, err := payloadAs[idRequest](req.Payload)
|
||||
if err != nil {
|
||||
return invalidPayloadResponse(req, err)
|
||||
}
|
||||
|
||||
task, err := c.tasks.EnqueueTask(ctx, input.ID)
|
||||
if err != nil {
|
||||
return mapTaskError(req, err)
|
||||
}
|
||||
|
||||
taskMap, err := taskToMap(task)
|
||||
if err != nil {
|
||||
return ErrorResponse(req, "internal.error", "internal server error", true)
|
||||
}
|
||||
|
||||
return SuccessResponse(req, map[string]any{
|
||||
"id": task.ID,
|
||||
"status": task.Status,
|
||||
"task": taskMap,
|
||||
})
|
||||
}
|
||||
|
||||
func mapTaskError(req Envelope, err error) Envelope {
|
||||
switch {
|
||||
case errors.Is(err, workflow.ErrInvalidTaskInput):
|
||||
return ErrorResponse(req, "task.invalid_input", err.Error(), false)
|
||||
case errors.Is(err, workflow.ErrTaskCannotBeEnqueued):
|
||||
return ErrorResponse(req, "task.conflict", err.Error(), false)
|
||||
case errors.Is(err, pgx.ErrNoRows):
|
||||
return ErrorResponse(req, "task.not_found", "task not found", false)
|
||||
default:
|
||||
return ErrorResponse(req, "internal.error", "internal server error", true)
|
||||
}
|
||||
}
|
||||
|
||||
func invalidPayloadResponse(req Envelope, err error) Envelope {
|
||||
return ErrorResponse(req, "task.invalid_payload", "invalid payload: "+err.Error(), false)
|
||||
}
|
||||
|
||||
// payloadAs decodes an envelope payload into a typed request by round-tripping
|
||||
// through JSON instead of ad hoc map indexing.
|
||||
func payloadAs[T any](payload map[string]any) (T, error) {
|
||||
var out T
|
||||
b, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
return out, json.Unmarshal(b, &out)
|
||||
}
|
||||
|
||||
func taskToMap(task storage.Task) (map[string]any, error) {
|
||||
b, err := json.Marshal(task)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out map[string]any
|
||||
if err := json.Unmarshal(b, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func stringOrEmpty(value *string) string {
|
||||
if value == nil {
|
||||
return ""
|
||||
}
|
||||
return *value
|
||||
}
|
||||
448
services/core/internal/protosocket/tasks_test.go
Normal file
448
services/core/internal/protosocket/tasks_test.go
Normal file
|
|
@ -0,0 +1,448 @@
|
|||
package protosocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
toki "git.toki-labs.com/toki/proto-socket/go"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"google.golang.org/protobuf/types/known/structpb"
|
||||
"nhooyr.io/websocket"
|
||||
|
||||
apphttp "github.com/nomadcode/nomadcode-core/internal/http"
|
||||
"github.com/nomadcode/nomadcode-core/internal/notification"
|
||||
"github.com/nomadcode/nomadcode-core/internal/storage"
|
||||
"github.com/nomadcode/nomadcode-core/internal/workflow"
|
||||
)
|
||||
|
||||
type fakeTaskService struct {
|
||||
created workflow.CreateTaskInput
|
||||
createOut storage.Task
|
||||
createErr error
|
||||
|
||||
listLimit int32
|
||||
listOut []storage.Task
|
||||
listErr error
|
||||
|
||||
getID string
|
||||
getOut storage.Task
|
||||
getErr error
|
||||
|
||||
enqueueID string
|
||||
enqueueOut storage.Task
|
||||
enqueueErr error
|
||||
}
|
||||
|
||||
func (f *fakeTaskService) CreateTask(_ context.Context, in workflow.CreateTaskInput) (storage.Task, error) {
|
||||
f.created = in
|
||||
return f.createOut, f.createErr
|
||||
}
|
||||
|
||||
func (f *fakeTaskService) ListTasks(_ context.Context, limit int32) ([]storage.Task, error) {
|
||||
f.listLimit = limit
|
||||
return f.listOut, f.listErr
|
||||
}
|
||||
|
||||
func (f *fakeTaskService) GetTask(_ context.Context, id string) (storage.Task, error) {
|
||||
f.getID = id
|
||||
return f.getOut, f.getErr
|
||||
}
|
||||
|
||||
func (f *fakeTaskService) EnqueueTask(_ context.Context, id string) (storage.Task, error) {
|
||||
f.enqueueID = id
|
||||
return f.enqueueOut, f.enqueueErr
|
||||
}
|
||||
|
||||
func dispatchTo(fake TaskService, req Envelope) Envelope {
|
||||
d := NewDispatcher()
|
||||
NewTaskChannels(fake).Register(d)
|
||||
return d.Dispatch(context.Background(), req)
|
||||
}
|
||||
|
||||
func TestTaskCreateActionCallsWorkflowAndReturnsResponse(t *testing.T) {
|
||||
fake := &fakeTaskService{
|
||||
createOut: storage.Task{ID: "task-1", Status: "pending", Title: "My Task", Source: "client"},
|
||||
}
|
||||
|
||||
res := dispatchTo(fake, Envelope{
|
||||
ID: "r1",
|
||||
Type: "request",
|
||||
Channel: "task",
|
||||
Action: "task.create",
|
||||
Payload: map[string]any{
|
||||
"title": "My Task",
|
||||
"source": "client",
|
||||
"payload": map[string]any{},
|
||||
},
|
||||
})
|
||||
|
||||
if fake.created.Title != "My Task" || fake.created.Source != "client" {
|
||||
t.Fatalf("workflow not called with expected input: %+v", fake.created)
|
||||
}
|
||||
if res.Type != "response" {
|
||||
t.Fatalf("expected response type, got %q (err=%+v)", res.Type, res.Error)
|
||||
}
|
||||
if res.CorrelationID != "r1" {
|
||||
t.Errorf("expected correlation_id r1, got %q", res.CorrelationID)
|
||||
}
|
||||
if res.Payload["id"] != "task-1" || res.Payload["status"] != "pending" {
|
||||
t.Errorf("unexpected response payload: %+v", res.Payload)
|
||||
}
|
||||
if _, ok := res.Payload["task"].(map[string]any); !ok {
|
||||
t.Errorf("expected nested task map in response payload: %+v", res.Payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskListActionForwardsLimitToService(t *testing.T) {
|
||||
fake := &fakeTaskService{
|
||||
listOut: []storage.Task{{ID: "task-1"}, {ID: "task-2"}},
|
||||
}
|
||||
|
||||
res := dispatchTo(fake, Envelope{
|
||||
ID: "r2",
|
||||
Type: "request",
|
||||
Channel: "task",
|
||||
Action: "task.list",
|
||||
Payload: map[string]any{"limit": float64(5)},
|
||||
})
|
||||
|
||||
if fake.listLimit != 5 {
|
||||
t.Errorf("expected limit 5 forwarded to service, got %d", fake.listLimit)
|
||||
}
|
||||
if res.Type != "response" {
|
||||
t.Fatalf("expected response type, got %q", res.Type)
|
||||
}
|
||||
tasks, ok := res.Payload["tasks"].([]any)
|
||||
if !ok || len(tasks) != 2 {
|
||||
t.Errorf("expected 2 tasks in payload, got %+v", res.Payload["tasks"])
|
||||
}
|
||||
|
||||
// Omitted limit forwards 0; the workflow service owns default/bounding.
|
||||
resDefault := dispatchTo(fake, Envelope{ID: "r2b", Action: "task.list", Channel: "task", Payload: map[string]any{}})
|
||||
if fake.listLimit != 0 {
|
||||
t.Errorf("expected omitted limit to forward 0, got %d", fake.listLimit)
|
||||
}
|
||||
if resDefault.Type != "response" {
|
||||
t.Errorf("expected response type for default limit, got %q", resDefault.Type)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskGetActionMapsNotFoundError(t *testing.T) {
|
||||
fake := &fakeTaskService{getErr: pgx.ErrNoRows}
|
||||
|
||||
res := dispatchTo(fake, Envelope{
|
||||
ID: "r3",
|
||||
Action: "task.get",
|
||||
Channel: "task",
|
||||
Payload: map[string]any{"id": "missing"},
|
||||
})
|
||||
|
||||
if fake.getID != "missing" {
|
||||
t.Errorf("expected get called with 'missing', got %q", fake.getID)
|
||||
}
|
||||
if res.Type != "error" {
|
||||
t.Fatalf("expected error type, got %q", res.Type)
|
||||
}
|
||||
if res.Error == nil || res.Error.Code != "task.not_found" {
|
||||
t.Errorf("expected task.not_found, got %+v", res.Error)
|
||||
}
|
||||
if res.Error.Retryable {
|
||||
t.Error("expected not_found to be non-retryable")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskEnqueueActionMapsConflictError(t *testing.T) {
|
||||
fake := &fakeTaskService{enqueueErr: workflow.ErrTaskCannotBeEnqueued}
|
||||
|
||||
res := dispatchTo(fake, Envelope{
|
||||
ID: "r4",
|
||||
Action: "task.enqueue",
|
||||
Channel: "task",
|
||||
Payload: map[string]any{"id": "task-9"},
|
||||
})
|
||||
|
||||
if res.Type != "error" {
|
||||
t.Fatalf("expected error type, got %q", res.Type)
|
||||
}
|
||||
if res.Error == nil || res.Error.Code != "task.conflict" {
|
||||
t.Errorf("expected task.conflict, got %+v", res.Error)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskActionRejectsInvalidPayload(t *testing.T) {
|
||||
fake := &fakeTaskService{}
|
||||
|
||||
res := dispatchTo(fake, Envelope{
|
||||
ID: "r5",
|
||||
Action: "task.get",
|
||||
Channel: "task",
|
||||
Payload: map[string]any{"id": float64(123)}, // wrong type for string id
|
||||
})
|
||||
|
||||
if res.Type != "error" {
|
||||
t.Fatalf("expected error type, got %q", res.Type)
|
||||
}
|
||||
if res.Error == nil || res.Error.Code != "task.invalid_payload" {
|
||||
t.Errorf("expected task.invalid_payload, got %+v", res.Error)
|
||||
}
|
||||
if fake.getID != "" {
|
||||
t.Errorf("service should not be called on invalid payload, got %q", fake.getID)
|
||||
}
|
||||
}
|
||||
|
||||
type captureBroadcaster struct {
|
||||
env Envelope
|
||||
err error
|
||||
}
|
||||
|
||||
func (c *captureBroadcaster) BroadcastEnvelope(_ context.Context, env Envelope) error {
|
||||
c.env = env
|
||||
return c.err
|
||||
}
|
||||
|
||||
func TestHandleTaskEventBroadcastsStatusChangedEnvelope(t *testing.T) {
|
||||
cap := &captureBroadcaster{}
|
||||
sink := NewTaskEventBroadcaster(cap)
|
||||
|
||||
err := sink.HandleTaskEvent(context.Background(), notification.TaskEvent{
|
||||
Type: notification.TaskEventRunning,
|
||||
TaskID: "task-7",
|
||||
Title: "Run me",
|
||||
Status: "running",
|
||||
Message: "started",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
if cap.env.Type != "event" || cap.env.Channel != "task" || cap.env.Action != "task.status.changed" {
|
||||
t.Fatalf("unexpected envelope: type=%q channel=%q action=%q", cap.env.Type, cap.env.Channel, cap.env.Action)
|
||||
}
|
||||
if cap.env.Payload["id"] != "task-7" || cap.env.Payload["status"] != "running" {
|
||||
t.Errorf("unexpected event payload: %+v", cap.env.Payload)
|
||||
}
|
||||
if cap.env.Payload["type"] != string(notification.TaskEventRunning) {
|
||||
t.Errorf("expected type %q, got %v", notification.TaskEventRunning, cap.env.Payload["type"])
|
||||
}
|
||||
}
|
||||
|
||||
// --- route-level contract tests over the real router + proto-socket server ---
|
||||
|
||||
func newContractRouter(t *testing.T, fake TaskService) (*httptest.Server, func()) {
|
||||
t.Helper()
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
srv := NewServer(Config{HeartbeatIntervalSec: 0, HeartbeatWaitSec: 0}, logger)
|
||||
NewTaskChannels(fake).Register(srv.Dispatcher())
|
||||
|
||||
router := apphttp.NewRouter(
|
||||
apphttp.NewHandler(nil, nil, nil, logger),
|
||||
logger,
|
||||
apphttp.AuthConfig{Username: "nomadcode", Password: "secret"},
|
||||
srv,
|
||||
"/proto-socket",
|
||||
)
|
||||
ts := httptest.NewServer(router)
|
||||
return ts, func() {
|
||||
ts.Close()
|
||||
_ = srv.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func wsURL(ts *httptest.Server) string {
|
||||
return strings.Replace(ts.URL, "http", "ws", 1) + "/proto-socket"
|
||||
}
|
||||
|
||||
func authDialOptions() *websocket.DialOptions {
|
||||
header := http.Header{}
|
||||
header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte("nomadcode:secret")))
|
||||
return &websocket.DialOptions{HTTPHeader: header}
|
||||
}
|
||||
|
||||
func TestRouterProtoSocketTaskListRequiresAuth(t *testing.T) {
|
||||
fake := &fakeTaskService{listOut: []storage.Task{{ID: "task-1"}}}
|
||||
ts, cleanup := newContractRouter(t, fake)
|
||||
defer cleanup()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Unauthenticated dial must fail at the route.
|
||||
if conn, _, err := websocket.Dial(ctx, wsURL(ts), nil); err == nil {
|
||||
conn.Close(websocket.StatusNormalClosure, "")
|
||||
t.Fatal("expected unauthenticated websocket dial to fail")
|
||||
}
|
||||
|
||||
// Authenticated dial succeeds and task.list returns a response envelope.
|
||||
conn, _, err := websocket.Dial(ctx, wsURL(ts), authDialOptions())
|
||||
if err != nil {
|
||||
t.Fatalf("authenticated dial failed: %v", err)
|
||||
}
|
||||
defer conn.Close(websocket.StatusNormalClosure, "")
|
||||
|
||||
client := toki.NewWsClient(conn, 0, 0, ParserMap())
|
||||
defer client.Close()
|
||||
|
||||
reqStruct, err := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: "req-list",
|
||||
Type: "request",
|
||||
Channel: "task",
|
||||
Action: "task.list",
|
||||
Payload: map[string]any{"limit": float64(20)},
|
||||
}.ToStruct()
|
||||
if err != nil {
|
||||
t.Fatalf("failed to build request: %v", err)
|
||||
}
|
||||
|
||||
resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 2*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("send request failed: %v", err)
|
||||
}
|
||||
res, err := EnvelopeFromStruct(resStruct)
|
||||
if err != nil {
|
||||
t.Fatalf("parse response failed: %v", err)
|
||||
}
|
||||
if res.Type != "response" || res.CorrelationID != "req-list" {
|
||||
t.Fatalf("unexpected response: type=%q correlation=%q", res.Type, res.CorrelationID)
|
||||
}
|
||||
if _, ok := res.Payload["tasks"]; !ok {
|
||||
t.Errorf("expected tasks payload, got %+v", res.Payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProtoSocketTaskActionsMatchRestSemantics(t *testing.T) {
|
||||
fake := &fakeTaskService{
|
||||
createOut: storage.Task{ID: "task-c", Status: "pending"},
|
||||
getOut: storage.Task{ID: "task-g", Status: "running"},
|
||||
enqueueOut: storage.Task{ID: "task-e", Status: "queued"},
|
||||
}
|
||||
ts, cleanup := newContractRouter(t, fake)
|
||||
defer cleanup()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
conn, _, err := websocket.Dial(ctx, wsURL(ts), authDialOptions())
|
||||
if err != nil {
|
||||
t.Fatalf("dial failed: %v", err)
|
||||
}
|
||||
defer conn.Close(websocket.StatusNormalClosure, "")
|
||||
client := toki.NewWsClient(conn, 0, 0, ParserMap())
|
||||
defer client.Close()
|
||||
|
||||
send := func(action string, payload map[string]any) Envelope {
|
||||
reqStruct, err := Envelope{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
ID: "req-" + action,
|
||||
Type: "request",
|
||||
Channel: "task",
|
||||
Action: action,
|
||||
Payload: payload,
|
||||
}.ToStruct()
|
||||
if err != nil {
|
||||
t.Fatalf("build request %s: %v", action, err)
|
||||
}
|
||||
resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 2*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("send %s: %v", action, err)
|
||||
}
|
||||
res, err := EnvelopeFromStruct(resStruct)
|
||||
if err != nil {
|
||||
t.Fatalf("parse %s: %v", action, err)
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
create := send("task.create", map[string]any{"title": "T", "source": "client", "payload": map[string]any{}})
|
||||
if create.Type != "response" || create.Payload["id"] != "task-c" || create.Payload["status"] != "pending" {
|
||||
t.Errorf("unexpected create response: %+v (err=%+v)", create.Payload, create.Error)
|
||||
}
|
||||
|
||||
get := send("task.get", map[string]any{"id": "task-g"})
|
||||
if get.Type != "response" {
|
||||
t.Errorf("unexpected get response type: %q", get.Type)
|
||||
}
|
||||
if task, ok := get.Payload["task"].(map[string]any); !ok || task["status"] != "running" {
|
||||
t.Errorf("unexpected get task payload: %+v", get.Payload)
|
||||
}
|
||||
|
||||
enqueue := send("task.enqueue", map[string]any{"id": "task-e"})
|
||||
if enqueue.Type != "response" || enqueue.Payload["status"] != "queued" {
|
||||
t.Errorf("unexpected enqueue response: %+v", enqueue.Payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProtoSocketTaskErrorEnvelopeCodes(t *testing.T) {
|
||||
fake := &fakeTaskService{
|
||||
getErr: pgx.ErrNoRows,
|
||||
enqueueErr: workflow.ErrTaskCannotBeEnqueued,
|
||||
createErr: workflow.ErrInvalidTaskInput,
|
||||
}
|
||||
ts, cleanup := newContractRouter(t, fake)
|
||||
defer cleanup()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
conn, _, err := websocket.Dial(ctx, wsURL(ts), authDialOptions())
|
||||
if err != nil {
|
||||
t.Fatalf("dial failed: %v", err)
|
||||
}
|
||||
defer conn.Close(websocket.StatusNormalClosure, "")
|
||||
client := toki.NewWsClient(conn, 0, 0, ParserMap())
|
||||
defer client.Close()
|
||||
|
||||
send := func(action string, payload map[string]any) Envelope {
|
||||
reqStruct, _ := Envelope{ProtocolVersion: ProtocolVersion, ID: "e-" + action, Type: "request", Channel: "task", Action: action, Payload: payload}.ToStruct()
|
||||
resStruct, err := toki.SendRequestTyped[*structpb.Struct, *structpb.Struct](&client.Communicator, reqStruct, 2*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("send %s: %v", action, err)
|
||||
}
|
||||
res, _ := EnvelopeFromStruct(resStruct)
|
||||
return res
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
action string
|
||||
payload map[string]any
|
||||
code string
|
||||
}{
|
||||
{"task.get", map[string]any{"id": "x"}, "task.not_found"},
|
||||
{"task.enqueue", map[string]any{"id": "x"}, "task.conflict"},
|
||||
{"task.create", map[string]any{"title": "T", "source": "client"}, "task.invalid_input"},
|
||||
{"task.get", map[string]any{"id": float64(1)}, "task.invalid_payload"},
|
||||
{"task.unknown", map[string]any{}, "UNSUPPORTED_ACTION"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
res := send(tc.action, tc.payload)
|
||||
if res.Type != "error" {
|
||||
t.Errorf("%s: expected error type, got %q", tc.action, res.Type)
|
||||
continue
|
||||
}
|
||||
if res.Error == nil || res.Error.Code != tc.code {
|
||||
t.Errorf("%s: expected code %q, got %+v", tc.action, tc.code, res.Error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProtoSocketServerUsesConfiguredHeartbeat(t *testing.T) {
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
srv := NewServer(Config{HeartbeatIntervalSec: 1, HeartbeatWaitSec: 1}, logger)
|
||||
defer srv.Close()
|
||||
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
conn, _, err := websocket.Dial(ctx, strings.Replace(ts.URL, "http", "ws", 1), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial failed: %v", err)
|
||||
}
|
||||
conn.Close(websocket.StatusNormalClosure, "")
|
||||
}
|
||||
Loading…
Reference in a new issue