diff --git a/agent-roadmap/phase/foundation-alignment/milestones/socket-session-loop.md b/agent-roadmap/archive/phase/foundation-alignment/milestones/socket-session-loop.md similarity index 52% rename from agent-roadmap/phase/foundation-alignment/milestones/socket-session-loop.md rename to agent-roadmap/archive/phase/foundation-alignment/milestones/socket-session-loop.md index 7c2ffe1..6e53208 100644 --- a/agent-roadmap/phase/foundation-alignment/milestones/socket-session-loop.md +++ b/agent-roadmap/archive/phase/foundation-alignment/milestones/socket-session-loop.md @@ -11,7 +11,7 @@ Go API와 Flutter client 사이에 proto-socket 기반 최소 session loop를 ## 상태 -[진행중] +[완료] ## 구현 잠금 @@ -33,20 +33,25 @@ Go API와 Flutter client 사이에 proto-socket 기반 최소 session loop를 API와 Flutter client 사이의 최소 연결, 요청, 연결 상태 표현을 묶는다. - [x] [api-handshake] API가 ALT Hello request를 받고 Hello response를 반환한다. 검증: API와 client가 같은 contract typeName 기준으로 메시지를 주고받는다. -- [ ] [client-connect] Flutter client가 configured socket endpoint에 연결할 client layer를 갖는다. -- [ ] [request-loop] 최소 request-response 메시지가 API와 client 사이에서 왕복한다. -- [ ] [disconnect-state] client에서 연결/해제/오류 상태를 UI나 state layer에서 표현할 수 있다. 검증: heartbeat/disconnect는 proto-socket 기본 동작을 따른다. -- [ ] [socket-test] handshake loop를 검증하는 테스트 또는 반복 가능한 smoke 명령이 있다. 검증: API가 long-running worker 작업을 직접 실행하지 않고 `bin/test`, `bin/lint`가 통과한다. +- [x] [client-connect] Flutter client가 configured socket endpoint에 연결할 client layer를 갖는다. +- [x] [request-loop] 최소 request-response 메시지가 API와 client 사이에서 왕복한다. +- [x] [disconnect-state] client에서 연결/해제/오류 상태를 UI나 state layer에서 표현할 수 있다. 검증: heartbeat/disconnect는 proto-socket 기본 동작을 따른다. +- [x] [socket-test] handshake loop를 검증하는 테스트 또는 반복 가능한 smoke 명령이 있다. 검증: API가 long-running worker 작업을 직접 실행하지 않고 `bin/test`, `bin/lint`가 통과한다. +- [x] [web-runtime-smoke] Flutter Web 브라우저 런타임에서 Go API socket endpoint에 연결하고 Hello request-response가 왕복한다. 검증: 로컬 API 실행 중 Flutter Web에서 handshake 성공 상태를 확인한다. +- [x] [mobile-runtime-smoke] Android 또는 iOS 런타임에서 Go API socket endpoint에 연결하고 Hello request-response가 왕복한다. 검증: emulator/simulator endpoint 차이를 반영해 handshake 성공 상태를 확인한다. ## 완료 리뷰 -- 상태: 없음 -- 요청일: 없음 -- 완료 근거: 아직 기능 Task와 Task별 검증이 충족되지 않았다. +- 상태: 승인됨 +- 요청일: 2026-05-28 +- 완료 근거: API Hello handshake, Flutter `AltSocketClient.hello`, Riverpod socket state, dashboard status 표시가 구현되어 있다. +- 완료 근거: `go test ./internal/contracts ./internal/socket`, `flutter test test/integrations/socket/alt_socket_client_test.dart test/widget_test.dart`, `bin/test`, `bin/lint`가 통과했다. +- 완료 근거: `ssh toki@toki-labs.com` macOS host에서 `dart run test -p chrome test_runtime/socket_web_runtime_smoke_test.dart --timeout 30s`가 통과했다. +- 완료 근거: 같은 host의 Android emulator `emulator-5554`에서 `flutter test -d emulator-5554 integration_test/socket_runtime_smoke_test.dart --dart-define=ALT_SOCKET_HOST=10.0.2.2 --dart-define=ALT_SOCKET_PORT=13000 --dart-define=ALT_SOCKET_PATH=/socket`가 통과했다. - 리뷰 필요: - - [ ] 사용자가 완료 결과를 확인했다 - - [ ] archive 이동을 승인했다 -- 리뷰 코멘트: 없음 + - [x] 사용자가 완료 결과를 확인했다 + - [x] archive 이동을 승인했다 +- 리뷰 코멘트: 사용자 승인에 따라 `[완료]`로 전환하고 archive로 이동했다. ## 범위 제외 diff --git a/agent-roadmap/current.md b/agent-roadmap/current.md index 7056903..1e781df 100644 --- a/agent-roadmap/current.md +++ b/agent-roadmap/current.md @@ -9,10 +9,7 @@ ## 활성 Milestone -- [진행중] Socket Session Loop - - Phase: `agent-roadmap/phase/foundation-alignment/PHASE.md` - - 경로: `agent-roadmap/phase/foundation-alignment/milestones/socket-session-loop.md` -- [계획] Persistence and Worker Backbone +- [진행중] Persistence and Worker Backbone - Phase: `agent-roadmap/phase/foundation-alignment/PHASE.md` - 경로: `agent-roadmap/phase/foundation-alignment/milestones/persistence-worker-backbone.md` - [계획] Client Integration Clone Adoption diff --git a/agent-roadmap/phase/foundation-alignment/PHASE.md b/agent-roadmap/phase/foundation-alignment/PHASE.md index bf6f3f1..75fa83a 100644 --- a/agent-roadmap/phase/foundation-alignment/PHASE.md +++ b/agent-roadmap/phase/foundation-alignment/PHASE.md @@ -19,10 +19,10 @@ ALT가 agent-first monorepo 기준선 위에서 계약, proto-socket session loo - [완료] Contract and Codegen Baseline - 경로: `agent-roadmap/archive/phase/foundation-alignment/milestones/contract-codegen-baseline.md` - 요약: ALT application protobuf 계약과 Go/Dart 생성 흐름을 반복 가능하게 만든다. -- [진행중] Socket Session Loop - - 경로: `agent-roadmap/phase/foundation-alignment/milestones/socket-session-loop.md` +- [완료] Socket Session Loop + - 경로: `agent-roadmap/archive/phase/foundation-alignment/milestones/socket-session-loop.md` - 요약: proto-socket 기반 API와 Flutter client 사이의 최소 handshake/request-response loop를 닫는다. -- [계획] Persistence and Worker Backbone +- [진행중] Persistence and Worker Backbone - 경로: `agent-roadmap/phase/foundation-alignment/milestones/persistence-worker-backbone.md` - 요약: PostgreSQL/Redis 기반의 durable storage와 비동기 worker 실행 기준을 마련한다. diff --git a/agent-roadmap/phase/foundation-alignment/milestones/persistence-worker-backbone.md b/agent-roadmap/phase/foundation-alignment/milestones/persistence-worker-backbone.md index 4263148..62476a5 100644 --- a/agent-roadmap/phase/foundation-alignment/milestones/persistence-worker-backbone.md +++ b/agent-roadmap/phase/foundation-alignment/milestones/persistence-worker-backbone.md @@ -11,7 +11,7 @@ PostgreSQL과 Redis를 ALT의 운영 기본 인프라로 고정하고, worker가 ## 상태 -[계획] +[진행중] ## 구현 잠금 @@ -26,31 +26,24 @@ PostgreSQL과 Redis를 ALT의 운영 기본 인프라로 고정하고, worker가 - local infrastructure와 worker configuration 정합성 - storage boundary와 domain mapping -## 필수 기능 +## 기능 ### Epic: [worker-backbone] Persistence and worker backbone PostgreSQL, Redis, worker job model, storage boundary 기준을 묶는다. -- [ ] [db-migrations] PostgreSQL migration entrypoint가 있다. -- [ ] [typed-queries] DB query를 type-safe하게 다루는 생성/검증 흐름이 있다. +- [ ] [db-migrations] PostgreSQL migration entrypoint가 있다. 검증: local PostgreSQL/Redis를 실행한 뒤 worker가 필요한 설정을 읽을 수 있다. +- [ ] [typed-queries] DB query를 type-safe하게 다루는 생성/검증 흐름이 있다. 검증: migration/query generation 흐름이 문서화되어 있다. - [ ] [redis-boundary] Redis 사용 목적과 key prefix 기준이 worker 문서나 config에 정리되어 있다. -- [ ] [job-model] worker가 비동기 작업을 표현하고 실행할 기본 job model을 갖는다. +- [ ] [job-model] worker가 비동기 작업을 표현하고 실행할 기본 job model을 갖는다. 검증: worker job이 API 내부 구현에 의존하지 않는다. - [ ] [storage-ports] domain과 persistence 사이의 adapter/port 경계가 정해져 있다. -- [ ] [infra-check] local PostgreSQL/Redis health를 확인하는 반복 가능한 절차가 있다. - -## 완료 기준 - -- [ ] local PostgreSQL/Redis를 실행한 뒤 worker가 필요한 설정을 읽을 수 있다. -- [ ] migration/query generation 흐름이 문서화되어 있다. -- [ ] worker job이 API 내부 구현에 의존하지 않는다. -- [ ] `bin/test`, `bin/lint`, `bin/build`가 통과한다. +- [x] [infra-check] local PostgreSQL/Redis health를 확인하는 반복 가능한 절차가 있다. 검증: `bin/test`, `bin/lint`, `bin/build`가 통과한다. ## 완료 리뷰 - 상태: 없음 - 요청일: 없음 -- 완료 근거: 아직 필수 기능과 완료 기준이 충족되지 않았다. +- 완료 근거: 아직 기능 Task와 Task별 검증이 충족되지 않았다. - 리뷰 필요: - [ ] 사용자가 완료 결과를 확인했다 - [ ] archive 이동을 승인했다 @@ -67,6 +60,9 @@ PostgreSQL, Redis, worker job model, storage boundary 기준을 묶는다. - 관련 경로: `services/worker/`, `deployments/local/`, `packages/domain/`, `bin/` - 표준선(선택): PostgreSQL은 durable store, Redis는 보조 cache/queue/pubsub surface로 둔다. +- 환경 불일치: 현재 실행 컨테이너에는 Docker가 없고 기본 `localhost:5432/6379`가 닫혀 있어 local compose health를 직접 확인할 수 없다. +- field 환경 불일치: `ssh toki@toki-labs.com` macOS host에는 Docker와 `alt-postgres`가 있으나 해당 PostgreSQL은 role/db `alt` 기본값과 맞지 않고, host `localhost:6379` Redis는 열려 있지 않다. +- 표준선(환경): repo 기준 반복 검증은 `deployments/local/docker-compose.yml`의 PostgreSQL `alt/alt`와 Redis를 `bin/infra-check`로 확인한다. 외부/field 인프라는 `DATABASE_URL`, `REDIS_URL`을 명시해 별도 확인한다. - 선행 작업: Workspace Baseline - 후속 작업: Korea Daily Data Foundation - 확인 필요: 없음 diff --git a/agent-task/archive/2026/05/m-persistence-worker-backbone/01_storage_foundation/USER_REVIEW.md b/agent-task/archive/2026/05/m-persistence-worker-backbone/01_storage_foundation/USER_REVIEW.md new file mode 100644 index 0000000..6b449fc --- /dev/null +++ b/agent-task/archive/2026/05/m-persistence-worker-backbone/01_storage_foundation/USER_REVIEW.md @@ -0,0 +1,53 @@ +# User Review Required - m-persistence-worker-backbone/01_storage_foundation + +## 요청 일시 + +2026-05-28 + +## 상태 + +RESOLVED + +## 사유 + +- 유형: environment-blocked +- 현재 리뷰 회차: 2 +- 최종 판정: PASS +- 요약: repo-fixable follow-up 항목은 닫혔고, 사용자의 원격 기준 지시에 따라 `ssh toki@toki-labs.com`의 field 서비스(`code-server-postgres`, `code-server-redis`)와 `code-server` 컨테이너에서 worker migration을 검증해 차단을 해소했다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `agent-task/m-persistence-worker-backbone/01_storage_foundation/plan_cloud_G07_0.log` | `agent-task/m-persistence-worker-backbone/01_storage_foundation/code_review_cloud_G07_0.log` | FAIL | local infra/migration 검증 미완료, raw `DATABASE_URL` logging, generated check drift false pass | +| `agent-task/m-persistence-worker-backbone/01_storage_foundation/plan_cloud_G07_1.log` | `agent-task/m-persistence-worker-backbone/01_storage_foundation/code_review_cloud_G07_1.log` | FAIL | redaction과 generated check는 보완됐으나 local compose health 검증은 외부 환경 전제로 차단 | +| `agent-task/m-persistence-worker-backbone/01_storage_foundation/USER_REVIEW.md` | user decision | PASS/RESOLVED | 원격 field 서비스 기준으로 검증 대상을 바로잡고 health/migration 성공 확인 | + +## 차단 근거 + +- 문제: `[REVIEW_STORAGE-1]`의 local compose 기준 해석이 잘못되었고, 사용자는 원격 field 환경 기준 검증을 요구했다. +- 현재 archive plan: `agent-task/m-persistence-worker-backbone/01_storage_foundation/plan_cloud_G07_1.log` +- 현재 archive review: `agent-task/m-persistence-worker-backbone/01_storage_foundation/code_review_cloud_G07_1.log` +- 검증 명령: `ssh toki@toki-labs.com 'docker exec code-server-postgres pg_isready -U nomadcode -d nomadcode; docker exec code-server-redis redis-cli ping; docker exec -u abc code-server bash -lc "cd /config/workspace/alt/services/worker; go run ./cmd/alt-worker-migrate"'` +- 실제 출력: `code-server-postgres`는 accepting connections, `code-server-redis`는 `PONG`, migration command는 `database_url="postgres://nomadcode:xxxxx@code-server-postgres:5432/nomadcode?sslmode=disable"`로 redaction된 로그와 함께 성공. +- 차단 판단 근거: 해소됨. 기존 stop state는 검증 대상을 현재 컨테이너/local compose로 잘못 해석한 결과였으며, 원격 field 기준에서는 PostgreSQL/Redis health와 migration command가 성공했다. + +## 사용자 결정 필요 + +- [x] 자동 follow-up plan/review를 계속 진행한다. +- [ ] 계획을 재작성한다. +- [x] 테스트 환경, secret, 외부 서비스, SDK, 장비 조건을 준비한 뒤 재시도한다. +- [ ] 작업 범위를 줄이거나 보류/폐기한다. + +## 재개 조건 + +- 원격 field 서비스 기준 PostgreSQL/Redis health와 worker migration command 성공 출력이 확인되었다. + +## 다음 실행 힌트 + +- 완료 처리한다. 이후 runtime은 `m-persistence-worker-backbone` completion metadata를 받아 Milestone 상태 갱신 여부를 판단한다. + +## 종료 규칙 + +- 사용자가 이 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`를 작성한다. diff --git a/apps/client/integration_test/socket_runtime_smoke_test.dart b/apps/client/integration_test/socket_runtime_smoke_test.dart new file mode 100644 index 0000000..93ad58e --- /dev/null +++ b/apps/client/integration_test/socket_runtime_smoke_test.dart @@ -0,0 +1,26 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:integration_test/integration_test.dart'; + +import 'package:alt_client/src/integrations/socket/alt_socket_client.dart'; +import 'package:alt_client/src/integrations/socket/socket_endpoint.dart'; + +void main() { + IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + + testWidgets('connects to configured ALT API and completes hello handshake', ( + tester, + ) async { + const endpoint = AltSocketEndpoint(); + + final client = await AltSocketClient.connect(endpoint); + addTearDown(client.close); + + final response = await client.hello(timeout: const Duration(seconds: 5)); + + expect(response.serverName, equals('alt-api')); + expect(response.serverVersion, equals('dev')); + expect(response.altProtocolVersion, equals('alt.v1')); + expect(response.capabilities, contains('hello')); + expect(response.capabilities, contains('request-response')); + }); +} diff --git a/apps/client/lib/src/integrations/socket/socket_endpoint.dart b/apps/client/lib/src/integrations/socket/socket_endpoint.dart index 44666a4..61028df 100644 --- a/apps/client/lib/src/integrations/socket/socket_endpoint.dart +++ b/apps/client/lib/src/integrations/socket/socket_endpoint.dart @@ -1,12 +1,25 @@ class AltSocketEndpoint { + static const defaultHost = String.fromEnvironment( + 'ALT_SOCKET_HOST', + defaultValue: '127.0.0.1', + ); + static const defaultPort = int.fromEnvironment( + 'ALT_SOCKET_PORT', + defaultValue: 8080, + ); + static const defaultPath = String.fromEnvironment( + 'ALT_SOCKET_PATH', + defaultValue: '/socket', + ); + final String host; final int port; final String path; const AltSocketEndpoint({ - this.host = '127.0.0.1', - this.port = 8080, - this.path = '/socket', + this.host = defaultHost, + this.port = defaultPort, + this.path = defaultPath, }); @override diff --git a/apps/client/pubspec.lock b/apps/client/pubspec.lock index fa4d53a..46977c4 100644 --- a/apps/client/pubspec.lock +++ b/apps/client/pubspec.lock @@ -142,6 +142,11 @@ packages: description: flutter source: sdk version: "0.0.0" + flutter_driver: + dependency: transitive + description: flutter + source: sdk + version: "0.0.0" flutter_lints: dependency: "direct dev" description: @@ -176,6 +181,11 @@ packages: url: "https://pub.dev" source: hosted version: "4.0.0" + fuchsia_remote_debug_protocol: + dependency: transitive + description: flutter + source: sdk + version: "0.0.0" glob: dependency: transitive description: @@ -208,6 +218,11 @@ packages: url: "https://pub.dev" source: hosted version: "4.1.2" + integration_test: + dependency: "direct dev" + description: flutter + source: sdk + version: "0.0.0" io: dependency: transitive description: @@ -312,6 +327,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.9.1" + platform: + dependency: transitive + description: + name: platform + sha256: "5d6b1b0036a5f331ebc77c850ebc8506cbc1e9416c27e59b439f917a902a4984" + url: "https://pub.dev" + source: hosted + version: "3.1.6" pool: dependency: transitive description: @@ -320,6 +343,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.5.2" + process: + dependency: transitive + description: + name: process + sha256: c6248e4526673988586e8c00bb22a49210c258dc91df5227d5da9748ecf79744 + url: "https://pub.dev" + source: hosted + version: "5.0.5" proto_socket: dependency: "direct main" description: @@ -452,6 +483,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.4.1" + sync_http: + dependency: transitive + description: + name: sync_http + sha256: "7f0cd72eca000d2e026bcd6f990b81d0ca06022ef4e32fb257b30d3d1014a961" + url: "https://pub.dev" + source: hosted + version: "0.3.1" term_glyph: dependency: transitive description: @@ -540,6 +579,14 @@ packages: url: "https://pub.dev" source: hosted version: "3.0.3" + webdriver: + dependency: transitive + description: + name: webdriver + sha256: "2f3a14ca026957870cfd9c635b83507e0e51d8091568e90129fbf805aba7cade" + url: "https://pub.dev" + source: hosted + version: "3.1.0" webkit_inspection_protocol: dependency: transitive description: diff --git a/apps/client/pubspec.yaml b/apps/client/pubspec.yaml index a32ae30..66ba128 100644 --- a/apps/client/pubspec.yaml +++ b/apps/client/pubspec.yaml @@ -44,6 +44,8 @@ dependencies: dev_dependencies: flutter_test: sdk: flutter + integration_test: + sdk: flutter # The "flutter_lints" package below contains a set of recommended lints to # encourage good coding practices. The lint set provided by the package is diff --git a/apps/client/test_runtime/socket_web_runtime_smoke_test.dart b/apps/client/test_runtime/socket_web_runtime_smoke_test.dart new file mode 100644 index 0000000..d259af6 --- /dev/null +++ b/apps/client/test_runtime/socket_web_runtime_smoke_test.dart @@ -0,0 +1,39 @@ +// ignore_for_file: depend_on_referenced_packages + +@TestOn('browser') +library; + +import 'package:test/test.dart'; + +import 'package:alt_client/src/integrations/socket/alt_socket_client.dart'; +import 'package:alt_client/src/integrations/socket/socket_endpoint.dart'; + +void main() { + test( + 'connects from Chrome runtime to ALT API and completes hello handshake', + () async { + const endpoint = AltSocketEndpoint( + host: String.fromEnvironment( + 'ALT_SOCKET_HOST', + defaultValue: '127.0.0.1', + ), + port: int.fromEnvironment('ALT_SOCKET_PORT', defaultValue: 13000), + path: String.fromEnvironment( + 'ALT_SOCKET_PATH', + defaultValue: '/socket', + ), + ); + + final client = await AltSocketClient.connect(endpoint); + addTearDown(client.close); + + final response = await client.hello(timeout: const Duration(seconds: 5)); + + expect(response.serverName, equals('alt-api')); + expect(response.serverVersion, equals('dev')); + expect(response.altProtocolVersion, equals('alt.v1')); + expect(response.capabilities, contains('hello')); + expect(response.capabilities, contains('request-response')); + }, + ); +} diff --git a/bin/dev b/bin/dev index c4a4cf2..52e8949 100755 --- a/bin/dev +++ b/bin/dev @@ -6,6 +6,7 @@ ALT development entrypoints: Local infra: docker compose -f deployments/local/docker-compose.yml up -d + bin/infra-check API: cd services/api && go run ./cmd/alt-api diff --git a/bin/infra-check b/bin/infra-check new file mode 100755 index 0000000..0061e24 --- /dev/null +++ b/bin/infra-check @@ -0,0 +1,65 @@ +#!/usr/bin/env bash +set -euo pipefail + +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +compose_file="$root/deployments/local/docker-compose.yml" + +database_url="${DATABASE_URL:-postgres://alt:alt@localhost:5432/alt?sslmode=disable}" +redis_url="${REDIS_URL:-redis://localhost:6379/0}" + +docker_cmd="" +if command -v docker >/dev/null 2>&1; then + docker_cmd="$(command -v docker)" +elif [ -x "/Applications/Docker.app/Contents/Resources/bin/docker" ]; then + docker_cmd="/Applications/Docker.app/Contents/Resources/bin/docker" +fi + +check_default_postgres() { + if [ -z "$docker_cmd" ]; then + echo "postgres: docker is unavailable; cannot inspect local compose service" >&2 + return 1 + fi + "$docker_cmd" compose -f "$compose_file" exec -T postgres pg_isready -U alt -d alt +} + +check_default_redis() { + if [ -z "$docker_cmd" ]; then + echo "redis: docker is unavailable; cannot inspect local compose service" >&2 + return 1 + fi + test "$("$docker_cmd" compose -f "$compose_file" exec -T redis redis-cli ping)" = "PONG" +} + +check_external_postgres() { + if command -v pg_isready >/dev/null 2>&1; then + pg_isready -d "$database_url" + return + fi + echo "postgres: pg_isready is required when DATABASE_URL overrides local compose" >&2 + return 1 +} + +check_external_redis() { + if command -v redis-cli >/dev/null 2>&1; then + test "$(redis-cli -u "$redis_url" ping)" = "PONG" + return + fi + echo "redis: redis-cli is required when REDIS_URL overrides local compose" >&2 + return 1 +} + +echo "ALT infrastructure health" + +if [ "$database_url" = "postgres://alt:alt@localhost:5432/alt?sslmode=disable" ]; then + check_default_postgres +else + check_external_postgres +fi +echo "postgres: ok" + +if [ "$redis_url" = "redis://localhost:6379/0" ]; then + check_default_redis +else + check_external_redis +fi +echo "redis: ok" diff --git a/bin/worker-storage-check b/bin/worker-storage-check new file mode 100755 index 0000000..79944a7 --- /dev/null +++ b/bin/worker-storage-check @@ -0,0 +1,30 @@ +#!/usr/bin/env bash +set -euo pipefail + +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +sqlc_out="$root/services/worker/internal/storage/postgres/sqlc" + +snapshot_dir="$(mktemp -d)" +trap 'rm -rf "$snapshot_dir"' EXIT + +snapshot_sqlc="$snapshot_dir/sqlc" +mkdir -p "$snapshot_sqlc" + +if [ -d "$sqlc_out" ]; then + cp -R "$sqlc_out/." "$snapshot_sqlc/" +fi + +"$root/bin/worker-storage-gen" + +drift=0 + +if ! diff -ru "$snapshot_sqlc" "$sqlc_out" >/dev/null; then + echo "worker-storage-check: Generated sqlc output drift detected under $sqlc_out" >&2 + diff -ru "$snapshot_sqlc" "$sqlc_out" >&2 || true + drift=1 +fi + +if [ "$drift" -ne 0 ]; then + echo "worker-storage-check: run bin/worker-storage-gen and commit the regenerated output" >&2 + exit 1 +fi diff --git a/bin/worker-storage-gen b/bin/worker-storage-gen new file mode 100755 index 0000000..79cd667 --- /dev/null +++ b/bin/worker-storage-gen @@ -0,0 +1,8 @@ +#!/usr/bin/env bash +set -euo pipefail + +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + +echo "Generating worker storage code via sqlc..." +(cd "$root/services/worker" && go run github.com/sqlc-dev/sqlc/cmd/sqlc generate) +echo "Generation complete." diff --git a/go.work.sum b/go.work.sum index 2c32801..e587130 100644 --- a/go.work.sum +++ b/go.work.sum @@ -1 +1,2 @@ -github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= diff --git a/services/api/internal/config/config.go b/services/api/internal/config/config.go index 3a68fe8..bfa03e7 100644 --- a/services/api/internal/config/config.go +++ b/services/api/internal/config/config.go @@ -3,6 +3,7 @@ package config import ( "os" "strconv" + "strings" ) type Config struct { @@ -11,6 +12,7 @@ type Config struct { SocketPath string HeartbeatIntervalSec int HeartbeatWaitSec int + WSOriginPatterns []string } func Load() Config { @@ -20,6 +22,7 @@ func Load() Config { SocketPath: getenv("ALT_API_SOCKET_PATH", "/socket"), HeartbeatIntervalSec: getenvInt("ALT_SOCKET_HEARTBEAT_INTERVAL_SEC", 30), HeartbeatWaitSec: getenvInt("ALT_SOCKET_HEARTBEAT_WAIT_SEC", 10), + WSOriginPatterns: getenvList("ALT_API_WS_ORIGIN_PATTERNS", "localhost:*,127.0.0.1:*"), } } @@ -41,3 +44,16 @@ func getenvInt(key string, fallback int) int { } return parsed } + +func getenvList(key string, fallback string) []string { + value := getenv(key, fallback) + parts := strings.Split(value, ",") + items := make([]string, 0, len(parts)) + for _, part := range parts { + item := strings.TrimSpace(part) + if item != "" { + items = append(items, item) + } + } + return items +} diff --git a/services/api/internal/config/config_test.go b/services/api/internal/config/config_test.go new file mode 100644 index 0000000..c25eec2 --- /dev/null +++ b/services/api/internal/config/config_test.go @@ -0,0 +1,33 @@ +package config + +import ( + "reflect" + "testing" +) + +func TestLoadDefaultsIncludeLocalWebSocketOrigins(t *testing.T) { + t.Setenv("ALT_API_HOST", "") + t.Setenv("ALT_API_PORT", "") + t.Setenv("ALT_API_SOCKET_PATH", "") + t.Setenv("ALT_SOCKET_HEARTBEAT_INTERVAL_SEC", "") + t.Setenv("ALT_SOCKET_HEARTBEAT_WAIT_SEC", "") + t.Setenv("ALT_API_WS_ORIGIN_PATTERNS", "") + + cfg := Load() + + want := []string{"localhost:*", "127.0.0.1:*"} + if !reflect.DeepEqual(cfg.WSOriginPatterns, want) { + t.Fatalf("WSOriginPatterns = %v, want %v", cfg.WSOriginPatterns, want) + } +} + +func TestLoadParsesWebSocketOriginPatterns(t *testing.T) { + t.Setenv("ALT_API_WS_ORIGIN_PATTERNS", "localhost:*, 127.0.0.1:* , toki-labs.com:*") + + cfg := Load() + + want := []string{"localhost:*", "127.0.0.1:*", "toki-labs.com:*"} + if !reflect.DeepEqual(cfg.WSOriginPatterns, want) { + t.Fatalf("WSOriginPatterns = %v, want %v", cfg.WSOriginPatterns, want) + } +} diff --git a/services/api/internal/socket/server.go b/services/api/internal/socket/server.go index a84c680..408362c 100644 --- a/services/api/internal/socket/server.go +++ b/services/api/internal/socket/server.go @@ -16,7 +16,13 @@ const ( ) func NewServer(cfg config.Config) *protoSocket.WsServer { - server := protoSocket.NewWsServer(cfg.Host, cfg.Port, cfg.SocketPath, func(conn *websocket.Conn) *protoSocket.WsClient { + options := protoSocket.WsServerOptions{} + if len(cfg.WSOriginPatterns) > 0 { + options.AcceptOptions = &websocket.AcceptOptions{ + OriginPatterns: cfg.WSOriginPatterns, + } + } + server := protoSocket.NewWsServerWithOptions(cfg.Host, cfg.Port, cfg.SocketPath, options, func(conn *websocket.Conn) *protoSocket.WsClient { return protoSocket.NewWsClient(conn, cfg.HeartbeatIntervalSec, cfg.HeartbeatWaitSec, apiContracts.ParserMap()) }) server.OnClientConnected = registerSessionHandlers diff --git a/services/worker/README.md b/services/worker/README.md new file mode 100644 index 0000000..29b6b55 --- /dev/null +++ b/services/worker/README.md @@ -0,0 +1,33 @@ +# ALT Worker Service + +The Worker service handles background tasks, such as market data imports, data normalization, and backtest execution. + +## Redis Usage Guidelines + +### Purpose +Redis is used as a lightweight, high-performance messaging, coordination, and caching layer for background jobs. It is **not** a source of truth. All critical system states and historical datasets reside in PostgreSQL. + +### Key Prefixing Standard +To avoid key collisions and maintain organization, all keys created by the worker **must** use the designated helper package `internal/rediskeys` and strictly follow the format: + +``` +:worker:: +``` + +- ``: Configured via `ALT_REDIS_KEY_PREFIX` (defaults to `alt`). This segregates environments or local dev workspaces. +- `worker`: Hardcoded namespace denoting worker-owned keys. +- ``: Lowercase segment describing the domain or usage (e.g. `job`, `lock`, `cache`). +- ``: Unique identifier for the specific object (e.g., job ID, lock name). + +Always utilize `rediskeys.WorkerKey(prefix, purpose, id)` instead of assembling key strings manually. + +## Configuration + +Environment variables can be used to customize worker runtime: + +| Variable | Description | Default | +|---|---|---| +| `DATABASE_URL` | PostgreSQL connection string | `postgres://alt:alt@localhost:5432/alt?sslmode=disable` | +| `REDIS_URL` | Redis connection string | `redis://localhost:6379/0` | +| `ALT_REDIS_KEY_PREFIX` | Namespace prefix for Redis keys | `alt` | +| `ALT_WORKER_QUEUE` | Target worker queue to consume from | `default` | diff --git a/services/worker/cmd/alt-worker-migrate/main.go b/services/worker/cmd/alt-worker-migrate/main.go new file mode 100644 index 0000000..9db5b5f --- /dev/null +++ b/services/worker/cmd/alt-worker-migrate/main.go @@ -0,0 +1,52 @@ +package main + +import ( + "context" + "log/slog" + "net/url" + "os" + "time" + + "github.com/jackc/pgx/v5" + + "git.toki-labs.com/toki/alt/services/worker/internal/config" + "git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres" +) + +// redactDatabaseURL parses and redacts the password from a database URL connection string. +func redactDatabaseURL(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil { + return "[invalid database url]" + } + if u.User != nil { + if _, hasPassword := u.User.Password(); hasPassword { + u.User = url.UserPassword(u.User.Username(), "xxxxx") + } + } + return u.String() +} + +func main() { + cfg := config.Load() + + slog.Info("starting migrations", "database_url", redactDatabaseURL(cfg.DatabaseURL)) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + conn, err := pgx.Connect(ctx, cfg.DatabaseURL) + if err != nil { + slog.Error("failed to connect to database", "error", err) + os.Exit(1) + } + defer conn.Close(ctx) + + err = postgres.RunMigrations(ctx, conn) + if err != nil { + slog.Error("failed to run migrations", "error", err) + os.Exit(1) + } + + slog.Info("migrations completed successfully") +} diff --git a/services/worker/cmd/alt-worker-migrate/main_test.go b/services/worker/cmd/alt-worker-migrate/main_test.go new file mode 100644 index 0000000..a2fa194 --- /dev/null +++ b/services/worker/cmd/alt-worker-migrate/main_test.go @@ -0,0 +1,30 @@ +package main + +import ( + "strings" + "testing" +) + +func TestDatabaseURLLogMetadataRedactsPassword(t *testing.T) { + rawURL := "postgres://alt:alt_secret_password@localhost:5432/alt?sslmode=disable" + redacted := redactDatabaseURL(rawURL) + + if strings.Contains(redacted, "alt_secret_password") { + t.Errorf("expected password to be redacted, but redacted string still contains password: %s", redacted) + } + + expected := "postgres://alt:xxxxx@localhost:5432/alt?sslmode=disable" + if redacted != expected { + t.Errorf("redacted URL mismatch: got %q, want %q", redacted, expected) + } +} + +func TestDatabaseURLLogMetadataRejectsRawSecret(t *testing.T) { + // Handles invalid URL format gracefully + invalidURL := "postgres://a b" // spaces are invalid in url.Parse + redacted := redactDatabaseURL(invalidURL) + + if redacted != "[invalid database url]" { + t.Errorf("expected invalid URL to be handled gracefully, got %q", redacted) + } +} diff --git a/services/worker/cmd/alt-worker/main.go b/services/worker/cmd/alt-worker/main.go index 5cb7509..48d2277 100644 --- a/services/worker/cmd/alt-worker/main.go +++ b/services/worker/cmd/alt-worker/main.go @@ -4,9 +4,16 @@ import ( "log/slog" "git.toki-labs.com/toki/alt/services/worker/internal/config" + "git.toki-labs.com/toki/alt/services/worker/internal/jobs" ) func main() { cfg := config.Load() - slog.Info("worker scaffold ready", "database_url_set", cfg.DatabaseURL != "", "redis_url_set", cfg.RedisURL != "") + runner := jobs.NewRunner() + jobs.RegisterBuiltins(runner) + slog.Info("worker ready", + "redis_key_prefix", cfg.RedisKeyPrefix, + "worker_queue", cfg.WorkerQueue, + "handlers", runner.Len(), + ) } diff --git a/services/worker/go.mod b/services/worker/go.mod index 8eda1f2..689671b 100644 --- a/services/worker/go.mod +++ b/services/worker/go.mod @@ -1,3 +1,64 @@ module git.toki-labs.com/toki/alt/services/worker go 1.22 + +require ( + github.com/jackc/pgx/v5 v5.7.2 + github.com/sqlc-dev/sqlc v1.27.0 +) + +require ( + filippo.io/edwards25519 v1.1.0 // indirect + github.com/antlr4-go/antlr/v4 v4.13.1 // indirect + github.com/cubicdaiya/gonp v1.0.4 // indirect + github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/fatih/structtag v1.2.0 // indirect + github.com/go-sql-driver/mysql v1.8.1 // indirect + github.com/google/cel-go v0.21.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect + github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/jinzhu/inflection v1.0.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/pganalyze/pg_query_go/v5 v5.1.0 // indirect + github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63 // indirect + github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect + github.com/pingcap/log v1.1.0 // indirect + github.com/pingcap/tidb/pkg/parser v0.0.0-20231103154709-4f00ece106b1 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/riza-io/grpc-go v0.2.0 // indirect + github.com/spf13/cobra v1.8.1 // indirect + github.com/spf13/pflag v1.0.5 // indirect + github.com/stoewer/go-strcase v1.2.0 // indirect + github.com/tetratelabs/wazero v1.7.3 // indirect + github.com/wasilibs/go-pgquery v0.0.0-20240606042535-c0843d6592cc // indirect + github.com/wasilibs/wazero-helpers v0.0.0-20240604052452-61d7981e9a38 // indirect + go.uber.org/atomic v1.11.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.26.0 // indirect + golang.org/x/crypto v0.31.0 // indirect + golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect + golang.org/x/net v0.25.0 // indirect + golang.org/x/sync v0.10.0 // indirect + golang.org/x/sys v0.28.0 // indirect + golang.org/x/text v0.21.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect + google.golang.org/grpc v1.65.0 // indirect + google.golang.org/protobuf v1.34.2 // indirect + gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect + modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect + modernc.org/libc v1.55.3 // indirect + modernc.org/mathutil v1.6.0 // indirect + modernc.org/memory v1.8.0 // indirect + modernc.org/sqlite v1.31.1 // indirect + modernc.org/strutil v1.2.0 // indirect + modernc.org/token v1.1.0 // indirect +) diff --git a/services/worker/go.sum b/services/worker/go.sum new file mode 100644 index 0000000..81c340a --- /dev/null +++ b/services/worker/go.sum @@ -0,0 +1,195 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= +github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= +github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= +github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o= +github.com/cubicdaiya/gonp v1.0.4 h1:ky2uIAJh81WiLcGKBVD5R7KsM/36W6IqqTy6Bo6rGws= +github.com/cubicdaiya/gonp v1.0.4/go.mod h1:iWGuP/7+JVTn02OWhRemVbMmG1DOUnmrGTYYACpOI0I= +github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 h1:iwZdTE0PVqJCos1vaoKsclOGD3ADKpshg3SRtYBbwso= +github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548/go.mod h1:e6NPNENfs9mPDVNRekM7lKScauxd5kXTr1Mfyig6TDM= +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/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/fatih/structtag v1.2.0 h1:/OdNE99OxoI/PqaW/SuSK9uxxT3f/tcSZgon/ssNSx4= +github.com/fatih/structtag v1.2.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94= +github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= +github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/google/cel-go v0.21.0 h1:cl6uW/gxN+Hy50tNYvI691+sXxioCnstFzLp2WO4GCI= +github.com/google/cel-go v0.21.0/go.mod h1:rHUlWCcBKgyEk+eV03RPdZUekPp6YcJwV0FxuUksYxc= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo= +github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= +github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.7.2 h1:mLoDLV6sonKlvjIEsV56SkWNCnuNv531l94GaIzO+XI= +github.com/jackc/pgx/v5 v5.7.2/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= +github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= +github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/pganalyze/pg_query_go/v5 v5.1.0 h1:MlxQqHZnvA3cbRQYyIrjxEjzo560P6MyTgtlaf3pmXg= +github.com/pganalyze/pg_query_go/v5 v5.1.0/go.mod h1:FsglvxidZsVN+Ltw3Ai6nTgPVcK2BPukH3jCDEqc1Ug= +github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63 h1:+FZIDR/D97YOPik4N4lPDaUcLDF/EQPogxtlHB2ZZRM= +github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg= +github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c h1:CgbKAHto5CQgWM9fSBIvaxsJHuGP0uM74HXtv3MyyGQ= +github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c/go.mod h1:4qGtCB0QK0wBzKtFEGDhxXnSnbQApw1gc9siScUl8ew= +github.com/pingcap/log v1.1.0 h1:ELiPxACz7vdo1qAvvaWJg1NrYFoY6gqAh/+Uo6aXdD8= +github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4= +github.com/pingcap/tidb/pkg/parser v0.0.0-20231103154709-4f00ece106b1 h1:SwGY3zMnK4wO85vvRIqrR3Yh6VpIC9pydG0QNOUPHCY= +github.com/pingcap/tidb/pkg/parser v0.0.0-20231103154709-4f00ece106b1/go.mod h1:yRkiqLFwIqibYg2P7h4bclHjHcJiIFRLKhGRyBcKYus= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/riza-io/grpc-go v0.2.0 h1:2HxQKFVE7VuYstcJ8zqpN84VnAoJ4dCL6YFhJewNcHQ= +github.com/riza-io/grpc-go v0.2.0/go.mod h1:2bDvR9KkKC3KhtlSHfR3dAXjUMT86kg4UfWFyVGWqi8= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM= +github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM= +github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y= +github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= +github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/sqlc-dev/sqlc v1.27.0 h1:wWc+401GLh0whLa30WmDkkl11lMBZuqvDvgu5OsaDiQ= +github.com/sqlc-dev/sqlc v1.27.0/go.mod h1:wXAlx++Ed1eUhMeEKyXfeCO+ogPIN1adG5DdPavR4k0= +github.com/stoewer/go-strcase v1.2.0 h1:Z2iHWqGXH00XYgqDmNgQbIBxf3wrNq0F3feEy0ainaU= +github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tetratelabs/wazero v1.7.3 h1:PBH5KVahrt3S2AHgEjKu4u+LlDbbk+nsGE3KLucy6Rw= +github.com/tetratelabs/wazero v1.7.3/go.mod h1:ytl6Zuh20R/eROuyDaGPkp82O9C/DJfXAwJfQ3X6/7Y= +github.com/wasilibs/go-pgquery v0.0.0-20240606042535-c0843d6592cc h1:Hgim1Xgk1+viV7p0aZh9OOrMRfG+E4mGA+JsI2uB0+k= +github.com/wasilibs/go-pgquery v0.0.0-20240606042535-c0843d6592cc/go.mod h1:ah6UfXIl/oA0K3SbourB/UHggVJOBXwPZ2XudDmmFac= +github.com/wasilibs/wazero-helpers v0.0.0-20240604052452-61d7981e9a38 h1:RBu75fhabyxyGJ2zhkoNuRyObBMhVeMoXqmeaPTg2CQ= +github.com/wasilibs/wazero-helpers v0.0.0-20240604052452-61d7981e9a38/go.mod h1:Z80JvMwvze8KUlVQIdw9L7OSskZJ1yxlpi4AQhoQe4s= +go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= +go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= +go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk= +go.uber.org/goleak v1.2.0/go.mod h1:XJYK+MuIchqpmGmUSAzotztawfKvYLUIgg7guXrwVUo= +go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/multierr v1.7.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI= +go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo= +go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= +golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= +golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 h1:vr/HnozRka3pE4EsMEg1lgkXJkTFJCVUX+S/ZT6wYzM= +golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842/go.mod h1:XtvwrStGgqGPLc4cjQfWqZHG1YFdYs6swckp8vpsjnc= +golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA= +golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac= +golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= +golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= +golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= +golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= +golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 h1:7whR9kGa5LUwFtpLm2ArCEejtnxlGeLbAyjFY8sGNFw= +google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157/go.mod h1:99sLkeliLXfdj2J75X3Ho+rrVCaJze0uwN7zDDkjPVU= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 h1:Zy9XzmMEflZ/MAaA7vNcoebnRAld7FsPW1EeBB7V0m8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157/go.mod h1:EfXuqaE1J41VCDicxHzUDm+8rk+7ZdXzHV0IhO/I6s0= +google.golang.org/grpc v1.65.0 h1:bs/cUb4lp1G5iImFFd3u5ixQzweKizoZJAwBNLR42lc= +google.golang.org/grpc v1.65.0/go.mod h1:WgYC2ypjlB0EiQi6wdKixMqukr6lBc0Vo+oOgjrM5ZQ= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/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/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k= +gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= +gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/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= +modernc.org/cc/v4 v4.21.4 h1:3Be/Rdo1fpr8GrQ7IVw9OHtplU4gWbb+wNgeoBMmGLQ= +modernc.org/cc/v4 v4.21.4/go.mod h1:HM7VJTZbUCR3rV8EYBi9wxnJ0ZBRiGE5OeGXNA0IsLQ= +modernc.org/ccgo/v4 v4.19.2 h1:lwQZgvboKD0jBwdaeVCTouxhxAyN6iawF3STraAal8Y= +modernc.org/ccgo/v4 v4.19.2/go.mod h1:ysS3mxiMV38XGRTTcgo0DQTeTmAO4oCmJl1nX9VFI3s= +modernc.org/fileutil v1.3.0 h1:gQ5SIzK3H9kdfai/5x41oQiKValumqNTDXMvKo62HvE= +modernc.org/fileutil v1.3.0/go.mod h1:XatxS8fZi3pS8/hKG2GH/ArUogfxjpEKs3Ku3aK4JyQ= +modernc.org/gc/v2 v2.4.1 h1:9cNzOqPyMJBvrUipmynX0ZohMhcxPtMccYgGOJdOiBw= +modernc.org/gc/v2 v2.4.1/go.mod h1:wzN5dK1AzVGoH6XOzc3YZ+ey/jPgYHLuVckd62P0GYU= +modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 h1:5D53IMaUuA5InSeMu9eJtlQXS2NxAhyWQvkKEgXZhHI= +modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6/go.mod h1:Qz0X07sNOR1jWYCrJMEnbW/X55x206Q7Vt4mz6/wHp4= +modernc.org/libc v1.55.3 h1:AzcW1mhlPNrRtjS5sS+eW2ISCgSOLLNyFzRh/V3Qj/U= +modernc.org/libc v1.55.3/go.mod h1:qFXepLhz+JjFThQ4kzwzOjA/y/artDeg+pcYnY+Q83w= +modernc.org/mathutil v1.6.0 h1:fRe9+AmYlaej+64JsEEhoWuAYBkOtQiMEU7n/XgfYi4= +modernc.org/mathutil v1.6.0/go.mod h1:Ui5Q9q1TR2gFm0AQRqQUaBWFLAhQpCwNcuhBOSedWPo= +modernc.org/memory v1.8.0 h1:IqGTL6eFMaDZZhEWwcREgeMXYwmW83LYW8cROZYkg+E= +modernc.org/memory v1.8.0/go.mod h1:XPZ936zp5OMKGWPqbD3JShgd/ZoQ7899TUuQqxY+peU= +modernc.org/opt v0.1.3 h1:3XOZf2yznlhC+ibLltsDGzABUGVx8J6pnFMS3E4dcq4= +modernc.org/opt v0.1.3/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0= +modernc.org/sortutil v1.2.0 h1:jQiD3PfS2REGJNzNCMMaLSp/wdMNieTbKX920Cqdgqc= +modernc.org/sortutil v1.2.0/go.mod h1:TKU2s7kJMf1AE84OoiGppNHJwvB753OYfNl2WRb++Ss= +modernc.org/sqlite v1.31.1 h1:XVU0VyzxrYHlBhIs1DiEgSl0ZtdnPtbLVy8hSkzxGrs= +modernc.org/sqlite v1.31.1/go.mod h1:UqoylwmTb9F+IqXERT8bW9zzOWN8qwAIcLdzeBZs4hA= +modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA= +modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/services/worker/internal/config/config.go b/services/worker/internal/config/config.go index 3224bf0..64c72cf 100644 --- a/services/worker/internal/config/config.go +++ b/services/worker/internal/config/config.go @@ -3,14 +3,18 @@ package config import "os" type Config struct { - DatabaseURL string - RedisURL string + DatabaseURL string + RedisURL string + RedisKeyPrefix string + WorkerQueue string } func Load() Config { return Config{ - DatabaseURL: getenv("DATABASE_URL", "postgres://alt:alt@localhost:5432/alt?sslmode=disable"), - RedisURL: getenv("REDIS_URL", "redis://localhost:6379/0"), + DatabaseURL: getenv("DATABASE_URL", "postgres://alt:alt@localhost:5432/alt?sslmode=disable"), + RedisURL: getenv("REDIS_URL", "redis://localhost:6379/0"), + RedisKeyPrefix: getenv("ALT_REDIS_KEY_PREFIX", "alt"), + WorkerQueue: getenv("ALT_WORKER_QUEUE", "default"), } } diff --git a/services/worker/internal/config/config_test.go b/services/worker/internal/config/config_test.go new file mode 100644 index 0000000..db11eaf --- /dev/null +++ b/services/worker/internal/config/config_test.go @@ -0,0 +1,77 @@ +package config + +import ( + "os" + "testing" +) + +func TestLoadDefaults(t *testing.T) { + // Ensure env vars are clean for default test + envs := []string{"DATABASE_URL", "REDIS_URL", "ALT_REDIS_KEY_PREFIX", "ALT_WORKER_QUEUE"} + for _, env := range envs { + if val, ok := os.LookupEnv(env); ok { + defer os.Setenv(env, val) + os.Unsetenv(env) + } else { + defer os.Unsetenv(env) + } + } + + cfg := Load() + + expectedDB := "postgres://alt:alt@localhost:5432/alt?sslmode=disable" + if cfg.DatabaseURL != expectedDB { + t.Errorf("expected DatabaseURL %q, got %q", expectedDB, cfg.DatabaseURL) + } + + expectedRedis := "redis://localhost:6379/0" + if cfg.RedisURL != expectedRedis { + t.Errorf("expected RedisURL %q, got %q", expectedRedis, cfg.RedisURL) + } + + expectedPrefix := "alt" + if cfg.RedisKeyPrefix != expectedPrefix { + t.Errorf("expected RedisKeyPrefix %q, got %q", expectedPrefix, cfg.RedisKeyPrefix) + } + + expectedQueue := "default" + if cfg.WorkerQueue != expectedQueue { + t.Errorf("expected WorkerQueue %q, got %q", expectedQueue, cfg.WorkerQueue) + } +} + +func TestLoadEnvOverrides(t *testing.T) { + envs := map[string]string{ + "DATABASE_URL": "postgres://user:pass@host:5432/db", + "REDIS_URL": "redis://redis-host:6379/1", + "ALT_REDIS_KEY_PREFIX": "custom-prefix", + "ALT_WORKER_QUEUE": "custom-queue", + } + + for k, v := range envs { + if val, ok := os.LookupEnv(k); ok { + defer os.Setenv(k, val) + } else { + defer os.Unsetenv(k) + } + os.Setenv(k, v) + } + + cfg := Load() + + if cfg.DatabaseURL != envs["DATABASE_URL"] { + t.Errorf("expected DatabaseURL %q, got %q", envs["DATABASE_URL"], cfg.DatabaseURL) + } + + if cfg.RedisURL != envs["REDIS_URL"] { + t.Errorf("expected RedisURL %q, got %q", envs["REDIS_URL"], cfg.RedisURL) + } + + if cfg.RedisKeyPrefix != envs["ALT_REDIS_KEY_PREFIX"] { + t.Errorf("expected RedisKeyPrefix %q, got %q", envs["ALT_REDIS_KEY_PREFIX"], cfg.RedisKeyPrefix) + } + + if cfg.WorkerQueue != envs["ALT_WORKER_QUEUE"] { + t.Errorf("expected WorkerQueue %q, got %q", envs["ALT_WORKER_QUEUE"], cfg.WorkerQueue) + } +} diff --git a/services/worker/internal/jobs/builtin.go b/services/worker/internal/jobs/builtin.go new file mode 100644 index 0000000..e173fc8 --- /dev/null +++ b/services/worker/internal/jobs/builtin.go @@ -0,0 +1,25 @@ +package jobs + +import ( + "context" + "encoding/json" + "log/slog" +) + +// RegisterBuiltins registers standard placeholder handlers for worker jobs. +func RegisterBuiltins(runner *Runner) { + runner.Register(KindImportDailyBars, func(ctx context.Context, payload json.RawMessage) error { + slog.Info("executing built-in job: import daily bars", "payload_len", len(payload)) + return nil + }) + + runner.Register(KindNormalizeDailyBars, func(ctx context.Context, payload json.RawMessage) error { + slog.Info("executing built-in job: normalize daily bars", "payload_len", len(payload)) + return nil + }) + + runner.Register(KindRunBacktest, func(ctx context.Context, payload json.RawMessage) error { + slog.Info("executing built-in job: run backtest", "payload_len", len(payload)) + return nil + }) +} diff --git a/services/worker/internal/jobs/job.go b/services/worker/internal/jobs/job.go new file mode 100644 index 0000000..72f72cf --- /dev/null +++ b/services/worker/internal/jobs/job.go @@ -0,0 +1,31 @@ +package jobs + +import "encoding/json" + +// Kind represents the type of worker job. +type Kind string + +const ( + KindImportDailyBars Kind = "import_daily_bars" + KindNormalizeDailyBars Kind = "normalize_daily_bars" + KindRunBacktest Kind = "run_backtest" +) + +// Status represents the state of a job. +type Status string + +const ( + StatusPending Status = "pending" + StatusRunning Status = "running" + StatusCompleted Status = "completed" + StatusFailed Status = "failed" +) + +// Job represents a background execution unit. +type Job struct { + ID string `json:"id"` + Kind Kind `json:"kind"` + Status Status `json:"status"` + Payload json.RawMessage `json:"payload,omitempty"` + Error string `json:"error,omitempty"` +} diff --git a/services/worker/internal/jobs/runner.go b/services/worker/internal/jobs/runner.go new file mode 100644 index 0000000..3b0f428 --- /dev/null +++ b/services/worker/internal/jobs/runner.go @@ -0,0 +1,66 @@ +package jobs + +import ( + "context" + "encoding/json" + "fmt" + "sync" +) + +// Handler defines the function signature for executing a job. +type Handler func(ctx context.Context, payload json.RawMessage) error + +// Runner manages the registration and execution of worker job handlers. +type Runner struct { + mu sync.RWMutex + handlers map[Kind]Handler +} + +// NewRunner creates a new Runner instance. +func NewRunner() *Runner { + return &Runner{ + handlers: make(map[Kind]Handler), + } +} + +// Register registers a handler for a specific job kind. +func (r *Runner) Register(kind Kind, handler Handler) { + r.mu.Lock() + defer r.mu.Unlock() + r.handlers[kind] = handler +} + +// Len returns the number of registered handlers. +func (r *Runner) Len() int { + r.mu.RLock() + defer r.mu.RUnlock() + return len(r.handlers) +} + +// Execute dispatches the job to its registered handler and executes it. +// It checks context cancellation and handles panics gracefully. +func (r *Runner) Execute(ctx context.Context, job Job) error { + r.mu.RLock() + handler, exists := r.handlers[job.Kind] + r.mu.RUnlock() + + if !exists { + return fmt.Errorf("unknown job kind: %s", job.Kind) + } + + // Check context before starting + if err := ctx.Err(); err != nil { + return err + } + + return r.executeWithPanicRecovery(ctx, handler, job.Payload) +} + +func (r *Runner) executeWithPanicRecovery(ctx context.Context, handler Handler, payload json.RawMessage) (err error) { + defer func() { + if rec := recover(); rec != nil { + err = fmt.Errorf("job panicked: %v", rec) + } + }() + return handler(ctx, payload) +} diff --git a/services/worker/internal/jobs/runner_test.go b/services/worker/internal/jobs/runner_test.go new file mode 100644 index 0000000..cfb1bb3 --- /dev/null +++ b/services/worker/internal/jobs/runner_test.go @@ -0,0 +1,110 @@ +package jobs + +import ( + "context" + "encoding/json" + "errors" + "testing" +) + +func TestRunnerDispatchesRegisteredHandler(t *testing.T) { + runner := NewRunner() + called := false + var receivedPayload json.RawMessage + + testKind := Kind("test_job") + runner.Register(testKind, func(ctx context.Context, payload json.RawMessage) error { + called = true + receivedPayload = payload + return nil + }) + + job := Job{ + ID: "job-1", + Kind: testKind, + Payload: json.RawMessage(`{"key":"value"}`), + } + + err := runner.Execute(context.Background(), job) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if !called { + t.Error("expected handler to be called, but it was not") + } + + if string(receivedPayload) != `{"key":"value"}` { + t.Errorf("expected payload %q, got %q", `{"key":"value"}`, string(receivedPayload)) + } +} + +func TestRunnerRejectsUnknownKind(t *testing.T) { + runner := NewRunner() + job := Job{ + ID: "job-2", + Kind: Kind("non_existent"), + } + + err := runner.Execute(context.Background(), job) + if err == nil { + t.Fatal("expected error for unknown job kind, got nil") + } +} + +func TestRunnerRespectsCanceledContext(t *testing.T) { + runner := NewRunner() + testKind := Kind("cancellation_test") + + runner.Register(testKind, func(ctx context.Context, payload json.RawMessage) error { + return nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel context immediately + + job := Job{ + ID: "job-3", + Kind: testKind, + } + + err := runner.Execute(ctx, job) + if err == nil { + t.Fatal("expected error due to canceled context, got nil") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("expected error %v, got %v", context.Canceled, err) + } +} + +func TestRunnerHandlesPanic(t *testing.T) { + runner := NewRunner() + panicKind := Kind("panic_job") + + runner.Register(panicKind, func(ctx context.Context, payload json.RawMessage) error { + panic("something went critically wrong") + }) + + job := Job{ + ID: "job-4", + Kind: panicKind, + } + + err := runner.Execute(context.Background(), job) + if err == nil { + t.Fatal("expected error from panic recovery, got nil") + } + if err.Error() != "job panicked: something went critically wrong" { + t.Errorf("unexpected error message: %v", err) + } +} + +func TestRegisterBuiltins(t *testing.T) { + runner := NewRunner() + RegisterBuiltins(runner) + + expectedCount := 3 + if runner.Len() != expectedCount { + t.Errorf("expected %d registered built-in handlers, got %d", expectedCount, runner.Len()) + } +} diff --git a/services/worker/internal/rediskeys/keys.go b/services/worker/internal/rediskeys/keys.go new file mode 100644 index 0000000..decd638 --- /dev/null +++ b/services/worker/internal/rediskeys/keys.go @@ -0,0 +1,31 @@ +package rediskeys + +import ( + "errors" + "fmt" + "strings" +) + +// WorkerKey builds a Redis key in the format ":worker::" +// prefix, purpose, and id are trimmed of whitespace and leading/trailing colons. +// If purpose or id is empty after normalization, an error is returned. +func WorkerKey(prefix, purpose, id string) (string, error) { + normPrefix := strings.Trim(prefix, " \t\n\r:") + if normPrefix == "" { + normPrefix = "alt" + } + normPrefix = strings.ToLower(normPrefix) + + normPurpose := strings.Trim(purpose, " \t\n\r:") + if normPurpose == "" { + return "", errors.New("rediskeys: purpose cannot be empty") + } + normPurpose = strings.ToLower(normPurpose) + + normID := strings.Trim(id, " \t\n\r:") + if normID == "" { + return "", errors.New("rediskeys: id cannot be empty") + } + + return fmt.Sprintf("%s:worker:%s:%s", normPrefix, normPurpose, normID), nil +} diff --git a/services/worker/internal/rediskeys/keys_test.go b/services/worker/internal/rediskeys/keys_test.go new file mode 100644 index 0000000..b29af86 --- /dev/null +++ b/services/worker/internal/rediskeys/keys_test.go @@ -0,0 +1,88 @@ +package rediskeys + +import ( + "strings" + "testing" +) + +func TestWorkerKey(t *testing.T) { + tests := []struct { + name string + prefix string + purpose string + id string + expectedKey string + expectError bool + }{ + { + name: "Standard key generation", + prefix: "alt", + purpose: "job", + id: "123", + expectedKey: "alt:worker:job:123", + expectError: false, + }, + { + name: "Case normalization", + prefix: "ALT", + purpose: "JOB", + id: "abc-DEF", + expectedKey: "alt:worker:job:abc-DEF", + expectError: false, + }, + { + name: "Whitespace and colon trimming", + prefix: " alt: ", + purpose: " :job: ", + id: " 123 ", + expectedKey: "alt:worker:job:123", + expectError: false, + }, + { + name: "Empty prefix fallback", + prefix: "", + purpose: "job", + id: "123", + expectedKey: "alt:worker:job:123", + expectError: false, + }, + { + name: "Empty purpose rejection", + prefix: "alt", + purpose: "", + id: "123", + expectedKey: "", + expectError: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + key, err := WorkerKey(tc.prefix, tc.purpose, tc.id) + if tc.expectError { + if err == nil { + t.Errorf("expected error, but got nil and key %q", key) + } + } else { + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if key != tc.expectedKey { + t.Errorf("expected key %q, got %q", tc.expectedKey, key) + } + } + }) + } +} + +func TestWorkerKeyRejectsEmptyID(t *testing.T) { + invalidIDs := []string{"", " ", "\t", " : : "} + for _, id := range invalidIDs { + key, err := WorkerKey("alt", "job", id) + if err == nil { + t.Errorf("expected error for empty/whitespace ID %q, but got nil and key %q", id, key) + } else if !strings.Contains(err.Error(), "id cannot be empty") { + t.Errorf("expected ID empty error message, got %v", err) + } + } +} diff --git a/services/worker/internal/storage/ports.go b/services/worker/internal/storage/ports.go new file mode 100644 index 0000000..c30291f --- /dev/null +++ b/services/worker/internal/storage/ports.go @@ -0,0 +1,25 @@ +package storage + +import ( + "context" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/backtest" + "git.toki-labs.com/toki/alt/packages/domain/market" +) + +type InstrumentStore interface { + UpsertInstrument(ctx context.Context, inst market.Instrument) error + GetInstrument(ctx context.Context, id market.InstrumentID) (market.Instrument, error) + ListInstruments(ctx context.Context) ([]market.Instrument, error) +} + +type BarStore interface { + UpsertBar(ctx context.Context, bar market.Bar) error + GetBars(ctx context.Context, id market.InstrumentID, timeframe market.Timeframe, from, to time.Time) ([]market.Bar, error) +} + +type BacktestRunStore interface { + UpsertRun(ctx context.Context, run backtest.Run) error + GetRun(ctx context.Context, id backtest.RunID) (backtest.Run, error) +} diff --git a/services/worker/internal/storage/postgres/mapping.go b/services/worker/internal/storage/postgres/mapping.go new file mode 100644 index 0000000..21738ee --- /dev/null +++ b/services/worker/internal/storage/postgres/mapping.go @@ -0,0 +1,204 @@ +package postgres + +import ( + "encoding/json" + "fmt" + "strconv" + + "github.com/jackc/pgx/v5/pgtype" + + "git.toki-labs.com/toki/alt/packages/domain/backtest" + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres/sqlc" +) + +func validateDecimal(val string) error { + if val == "" { + return fmt.Errorf("empty decimal value") + } + if _, err := strconv.ParseFloat(val, 64); err != nil { + return fmt.Errorf("invalid decimal format: %s", val) + } + return nil +} + +func stringToNumeric(val string) (pgtype.Numeric, error) { + var num pgtype.Numeric + if err := validateDecimal(val); err != nil { + return num, err + } + if err := num.Scan(val); err != nil { + return num, fmt.Errorf("failed to scan numeric: %w", err) + } + return num, nil +} + +func numericToString(num pgtype.Numeric) (string, error) { + if !num.Valid { + return "0", nil + } + val, err := num.Value() + if err != nil { + return "", err + } + switch v := val.(type) { + case string: + return v, nil + case []byte: + return string(v), nil + default: + return fmt.Sprintf("%v", v), nil + } +} + +func mapInstrumentToParams(inst market.Instrument) (sqlc.UpsertInstrumentParams, error) { + provBytes, err := json.Marshal(inst.ProviderSymbols) + if err != nil { + return sqlc.UpsertInstrumentParams{}, fmt.Errorf("failed to marshal provider symbols: %w", err) + } + if provBytes == nil { + provBytes = []byte("{}") + } + return sqlc.UpsertInstrumentParams{ + ID: string(inst.ID), + Market: string(inst.Market), + Venue: string(inst.Venue), + Symbol: inst.Symbol, + Name: inst.Name, + Currency: string(inst.Currency), + ProviderSymbols: provBytes, + }, nil +} + +func mapRowToInstrument(row sqlc.Instrument) (market.Instrument, error) { + var provSymbols map[string]string + if len(row.ProviderSymbols) > 0 { + if err := json.Unmarshal(row.ProviderSymbols, &provSymbols); err != nil { + return market.Instrument{}, fmt.Errorf("failed to unmarshal provider symbols: %w", err) + } + } + if provSymbols == nil { + provSymbols = make(map[string]string) + } + return market.Instrument{ + ID: market.InstrumentID(row.ID), + Market: market.Market(row.Market), + Venue: market.Venue(row.Venue), + Symbol: row.Symbol, + Name: row.Name, + Currency: market.Currency(row.Currency), + ProviderSymbols: provSymbols, + }, nil +} + +func mapBarToParams(bar market.Bar) (sqlc.UpsertBarParams, error) { + open, err := stringToNumeric(bar.Open.Amount.Value) + if err != nil { + return sqlc.UpsertBarParams{}, fmt.Errorf("invalid open amount: %w", err) + } + high, err := stringToNumeric(bar.High.Amount.Value) + if err != nil { + return sqlc.UpsertBarParams{}, fmt.Errorf("invalid high amount: %w", err) + } + low, err := stringToNumeric(bar.Low.Amount.Value) + if err != nil { + return sqlc.UpsertBarParams{}, fmt.Errorf("invalid low amount: %w", err) + } + closePrice, err := stringToNumeric(bar.Close.Amount.Value) + if err != nil { + return sqlc.UpsertBarParams{}, fmt.Errorf("invalid close amount: %w", err) + } + volume, err := stringToNumeric(bar.Volume.Amount.Value) + if err != nil { + return sqlc.UpsertBarParams{}, fmt.Errorf("invalid volume amount: %w", err) + } + + return sqlc.UpsertBarParams{ + InstrumentID: string(bar.InstrumentID), + Timeframe: string(bar.Timeframe), + Timestamp: pgtype.Timestamptz{Time: bar.Timestamp, Valid: !bar.Timestamp.IsZero()}, + Open: open, + High: high, + Low: low, + Close: closePrice, + Volume: volume, + }, nil +} + +func mapRowToBar(row sqlc.Bar, currency market.Currency) (market.Bar, error) { + openStr, err := numericToString(row.Open) + if err != nil { + return market.Bar{}, err + } + highStr, err := numericToString(row.High) + if err != nil { + return market.Bar{}, err + } + lowStr, err := numericToString(row.Low) + if err != nil { + return market.Bar{}, err + } + closeStr, err := numericToString(row.Close) + if err != nil { + return market.Bar{}, err + } + volumeStr, err := numericToString(row.Volume) + if err != nil { + return market.Bar{}, err + } + + return market.Bar{ + InstrumentID: market.InstrumentID(row.InstrumentID), + Timeframe: market.Timeframe(row.Timeframe), + Timestamp: row.Timestamp.Time, + Open: market.Price{ + Currency: currency, + Amount: market.Decimal{Value: openStr}, + }, + High: market.Price{ + Currency: currency, + Amount: market.Decimal{Value: highStr}, + }, + Low: market.Price{ + Currency: currency, + Amount: market.Decimal{Value: lowStr}, + }, + Close: market.Price{ + Currency: currency, + Amount: market.Decimal{Value: closeStr}, + }, + Volume: market.Quantity{ + Amount: market.Decimal{Value: volumeStr}, + }, + }, nil +} + +func mapRunToParams(run backtest.Run) (sqlc.UpsertRunParams, error) { + return sqlc.UpsertRunParams{ + ID: string(run.ID), + StrategyID: string(run.Spec.StrategyID), + Market: string(run.Spec.Market), + Timeframe: string(run.Spec.Timeframe), + FromTime: pgtype.Timestamptz{Time: run.Spec.From, Valid: !run.Spec.From.IsZero()}, + ToTime: pgtype.Timestamptz{Time: run.Spec.To, Valid: !run.Spec.To.IsZero()}, + Status: string(run.Status), + CreatedAt: pgtype.Timestamptz{Time: run.CreatedAt, Valid: !run.CreatedAt.IsZero()}, + UpdatedAt: pgtype.Timestamptz{Time: run.UpdatedAt, Valid: !run.UpdatedAt.IsZero()}, + }, nil +} + +func mapRowToRun(row sqlc.BacktestRun) (backtest.Run, error) { + return backtest.Run{ + ID: backtest.RunID(row.ID), + Spec: backtest.RunSpec{ + StrategyID: backtest.StrategyID(row.StrategyID), + Market: market.Market(row.Market), + Timeframe: market.Timeframe(row.Timeframe), + From: row.FromTime.Time, + To: row.ToTime.Time, + }, + Status: backtest.RunStatus(row.Status), + CreatedAt: row.CreatedAt.Time, + UpdatedAt: row.UpdatedAt.Time, + }, nil +} diff --git a/services/worker/internal/storage/postgres/mapping_test.go b/services/worker/internal/storage/postgres/mapping_test.go new file mode 100644 index 0000000..06ef349 --- /dev/null +++ b/services/worker/internal/storage/postgres/mapping_test.go @@ -0,0 +1,118 @@ +package postgres + +import ( + "testing" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/backtest" + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres/sqlc" +) + +func TestBacktestRunMappingRoundTrip(t *testing.T) { + now := time.Now().UTC().Truncate(time.Microsecond) // Truncate because postgres timestamptz resolves to microsecond precision + run := backtest.Run{ + ID: backtest.RunID("run-123"), + Spec: backtest.RunSpec{ + StrategyID: backtest.StrategyID("strat-abc"), + Market: market.MarketKR, + Timeframe: market.TimeframeDaily, + From: now.Add(-24 * time.Hour), + To: now, + }, + Status: backtest.RunStatusSucceeded, + CreatedAt: now.Add(-25 * time.Hour), + UpdatedAt: now, + } + + params, err := mapRunToParams(run) + if err != nil { + t.Fatalf("failed to map run to params: %v", err) + } + + row := sqlc.BacktestRun{ + ID: params.ID, + StrategyID: params.StrategyID, + Market: params.Market, + Timeframe: params.Timeframe, + FromTime: params.FromTime, + ToTime: params.ToTime, + Status: params.Status, + CreatedAt: params.CreatedAt, + UpdatedAt: params.UpdatedAt, + } + + mappedBack, err := mapRowToRun(row) + if err != nil { + t.Fatalf("failed to map row back to run: %v", err) + } + + if mappedBack.ID != run.ID { + t.Errorf("ID mismatch: got %v, want %v", mappedBack.ID, run.ID) + } + if mappedBack.Spec.StrategyID != run.Spec.StrategyID { + t.Errorf("StrategyID mismatch: got %v, want %v", mappedBack.Spec.StrategyID, run.Spec.StrategyID) + } + if mappedBack.Spec.Market != run.Spec.Market { + t.Errorf("Market mismatch: got %v, want %v", mappedBack.Spec.Market, run.Spec.Market) + } + if mappedBack.Spec.Timeframe != run.Spec.Timeframe { + t.Errorf("Timeframe mismatch: got %v, want %v", mappedBack.Spec.Timeframe, run.Spec.Timeframe) + } + if !mappedBack.Spec.From.Equal(run.Spec.From) { + t.Errorf("From mismatch: got %v, want %v", mappedBack.Spec.From, run.Spec.From) + } + if !mappedBack.Spec.To.Equal(run.Spec.To) { + t.Errorf("To mismatch: got %v, want %v", mappedBack.Spec.To, run.Spec.To) + } + if mappedBack.Status != run.Status { + t.Errorf("Status mismatch: got %v, want %v", mappedBack.Status, run.Status) + } + if !mappedBack.CreatedAt.Equal(run.CreatedAt) { + t.Errorf("CreatedAt mismatch: got %v, want %v", mappedBack.CreatedAt, run.CreatedAt) + } + if !mappedBack.UpdatedAt.Equal(run.UpdatedAt) { + t.Errorf("UpdatedAt mismatch: got %v, want %v", mappedBack.UpdatedAt, run.UpdatedAt) + } +} + +func TestMarketBarMappingRejectsInvalidDecimal(t *testing.T) { + now := time.Now().UTC() + bar := market.Bar{ + InstrumentID: market.InstrumentID("KRX:005930"), + Timeframe: market.TimeframeDaily, + Timestamp: now, + Open: market.Price{ + Currency: market.CurrencyKRW, + Amount: market.Decimal{Value: "invalid_decimal"}, + }, + High: market.Price{ + Currency: market.CurrencyKRW, + Amount: market.Decimal{Value: "50000"}, + }, + Low: market.Price{ + Currency: market.CurrencyKRW, + Amount: market.Decimal{Value: "49000"}, + }, + Close: market.Price{ + Currency: market.CurrencyKRW, + Amount: market.Decimal{Value: "49500"}, + }, + Volume: market.Quantity{ + Amount: market.Decimal{Value: "1000000"}, + }, + } + + _, err := mapBarToParams(bar) + if err == nil { + t.Fatal("expected error when mapping bar with invalid open decimal, got nil") + } + + // Make Open valid but Volume invalid + bar.Open.Amount.Value = "50000" + bar.Volume.Amount.Value = "abc" + _, err = mapBarToParams(bar) + if err == nil { + t.Fatal("expected error when mapping bar with invalid volume decimal, got nil") + } +} diff --git a/services/worker/internal/storage/postgres/migrate.go b/services/worker/internal/storage/postgres/migrate.go new file mode 100644 index 0000000..c68c81c --- /dev/null +++ b/services/worker/internal/storage/postgres/migrate.go @@ -0,0 +1,99 @@ +package postgres + +import ( + "context" + "embed" + "fmt" + "path/filepath" + "sort" + "strconv" + "strings" + + "github.com/jackc/pgx/v5" +) + +//go:embed migrations/*.sql +var migrationsFS embed.FS + +// RunMigrations runs all embedded migrations in order. +func RunMigrations(ctx context.Context, conn *pgx.Conn) error { + // Create schema_migrations table if not exists + _, err := conn.Exec(ctx, ` + CREATE TABLE IF NOT EXISTS schema_migrations ( + version BIGINT PRIMARY KEY, + applied_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP + ); + `) + if err != nil { + return fmt.Errorf("failed to create schema_migrations table: %w", err) + } + + entries, err := migrationsFS.ReadDir("migrations") + if err != nil { + return fmt.Errorf("failed to read migrations directory: %w", err) + } + + var migrationFiles []string + for _, entry := range entries { + if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".up.sql") { + migrationFiles = append(migrationFiles, entry.Name()) + } + } + sort.Strings(migrationFiles) + + if len(migrationFiles) == 0 { + return fmt.Errorf("no up migration files found") + } + + for _, filename := range migrationFiles { + parts := strings.Split(filename, "_") + if len(parts) == 0 { + return fmt.Errorf("invalid migration filename: %s", filename) + } + version, err := strconv.ParseInt(parts[0], 10, 64) + if err != nil { + return fmt.Errorf("failed to parse migration version from filename %s: %w", filename, err) + } + + // Check if version is already applied + var exists bool + err = conn.QueryRow(ctx, "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)", version).Scan(&exists) + if err != nil { + return fmt.Errorf("failed to check migration status for version %d: %w", version, err) + } + + if exists { + continue + } + + // Read migration content + content, err := migrationsFS.ReadFile(filepath.Join("migrations", filename)) + if err != nil { + return fmt.Errorf("failed to read migration file %s: %w", filename, err) + } + + // Run migration within transaction + tx, err := conn.Begin(ctx) + if err != nil { + return fmt.Errorf("failed to begin transaction for migration %s: %w", filename, err) + } + defer tx.Rollback(ctx) + + _, err = tx.Exec(ctx, string(content)) + if err != nil { + return fmt.Errorf("failed to execute migration %s: %w", filename, err) + } + + _, err = tx.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version) + if err != nil { + return fmt.Errorf("failed to record migration version %d: %w", version, err) + } + + err = tx.Commit(ctx) + if err != nil { + return fmt.Errorf("failed to commit transaction for migration %s: %w", filename, err) + } + } + + return nil +} diff --git a/services/worker/internal/storage/postgres/migrate_test.go b/services/worker/internal/storage/postgres/migrate_test.go new file mode 100644 index 0000000..2bb03ad --- /dev/null +++ b/services/worker/internal/storage/postgres/migrate_test.go @@ -0,0 +1,81 @@ +package postgres + +import ( + "strconv" + "strings" + "testing" +) + +func TestMigrationsAreEmbeddedInOrder(t *testing.T) { + entries, err := migrationsFS.ReadDir("migrations") + if err != nil { + t.Fatalf("failed to read migrations directory: %v", err) + } + + var upFiles []string + var downFiles []string + + for _, entry := range entries { + if entry.IsDir() { + continue + } + name := entry.Name() + if strings.HasSuffix(name, ".up.sql") { + upFiles = append(upFiles, name) + } else if strings.HasSuffix(name, ".down.sql") { + downFiles = append(downFiles, name) + } + } + + if len(upFiles) == 0 { + t.Error("expected at least one up migration file, found none") + } + if len(downFiles) == 0 { + t.Error("expected at least one down migration file, found none") + } + + if len(upFiles) != len(downFiles) { + t.Errorf("mismatched migration count: %d up files, %d down files", len(upFiles), len(downFiles)) + } + + // Verify up and down files match and are in numerical order + for i := 0; i < len(upFiles); i++ { + up := upFiles[i] + partsUp := strings.Split(up, "_") + if len(partsUp) < 2 { + t.Errorf("invalid up migration file name: %s", up) + continue + } + versionUp, err := strconv.ParseInt(partsUp[0], 10, 64) + if err != nil { + t.Errorf("failed to parse up version from %s: %v", up, err) + } + + if versionUp != int64(i+1) { + t.Errorf("expected version %d for file %s, got %d", i+1, up, versionUp) + } + + // Find corresponding down file + var foundDown bool + for _, down := range downFiles { + if strings.Split(down, "_")[0] == partsUp[0] { + foundDown = true + break + } + } + if !foundDown { + t.Errorf("missing corresponding down migration for: %s", up) + } + } +} + +func TestMigrationStatementsAreNamed(t *testing.T) { + // Check if migrate.go FS variable is populated + entries, err := migrationsFS.ReadDir("migrations") + if err != nil { + t.Fatalf("failed to read migrations: %v", err) + } + if len(entries) == 0 { + t.Error("migrations FS has no entries") + } +} diff --git a/services/worker/internal/storage/postgres/migrations/000001_worker_backbone.down.sql b/services/worker/internal/storage/postgres/migrations/000001_worker_backbone.down.sql new file mode 100644 index 0000000..3775348 --- /dev/null +++ b/services/worker/internal/storage/postgres/migrations/000001_worker_backbone.down.sql @@ -0,0 +1,3 @@ +DROP TABLE IF EXISTS backtest_runs; +DROP TABLE IF EXISTS bars; +DROP TABLE IF EXISTS instruments; diff --git a/services/worker/internal/storage/postgres/migrations/000001_worker_backbone.up.sql b/services/worker/internal/storage/postgres/migrations/000001_worker_backbone.up.sql new file mode 100644 index 0000000..af11f07 --- /dev/null +++ b/services/worker/internal/storage/postgres/migrations/000001_worker_backbone.up.sql @@ -0,0 +1,33 @@ +CREATE TABLE IF NOT EXISTS instruments ( + id TEXT PRIMARY KEY, + market TEXT NOT NULL, + venue TEXT NOT NULL, + symbol TEXT NOT NULL, + name TEXT NOT NULL, + currency TEXT NOT NULL, + provider_symbols JSONB NOT NULL +); + +CREATE TABLE IF NOT EXISTS bars ( + instrument_id TEXT NOT NULL REFERENCES instruments(id) ON DELETE CASCADE, + timeframe TEXT NOT NULL, + timestamp TIMESTAMPTZ NOT NULL, + open NUMERIC NOT NULL, + high NUMERIC NOT NULL, + low NUMERIC NOT NULL, + close NUMERIC NOT NULL, + volume NUMERIC NOT NULL, + PRIMARY KEY (instrument_id, timeframe, timestamp) +); + +CREATE TABLE IF NOT EXISTS backtest_runs ( + id TEXT PRIMARY KEY, + strategy_id TEXT NOT NULL, + market TEXT NOT NULL, + timeframe TEXT NOT NULL, + from_time TIMESTAMPTZ NOT NULL, + to_time TIMESTAMPTZ NOT NULL, + status TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL +); diff --git a/services/worker/internal/storage/postgres/queries/queries.sql b/services/worker/internal/storage/postgres/queries/queries.sql new file mode 100644 index 0000000..8b4062a --- /dev/null +++ b/services/worker/internal/storage/postgres/queries/queries.sql @@ -0,0 +1,53 @@ +-- name: UpsertInstrument :exec +INSERT INTO instruments (id, market, venue, symbol, name, currency, provider_symbols) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (id) DO UPDATE SET + market = EXCLUDED.market, + venue = EXCLUDED.venue, + symbol = EXCLUDED.symbol, + name = EXCLUDED.name, + currency = EXCLUDED.currency, + provider_symbols = EXCLUDED.provider_symbols; + +-- name: GetInstrument :one +SELECT id, market, venue, symbol, name, currency, provider_symbols +FROM instruments +WHERE id = $1; + +-- name: ListInstruments :many +SELECT id, market, venue, symbol, name, currency, provider_symbols +FROM instruments; + +-- name: UpsertBar :exec +INSERT INTO bars (instrument_id, timeframe, timestamp, open, high, low, close, volume) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +ON CONFLICT (instrument_id, timeframe, timestamp) DO UPDATE SET + open = EXCLUDED.open, + high = EXCLUDED.high, + low = EXCLUDED.low, + close = EXCLUDED.close, + volume = EXCLUDED.volume; + +-- name: GetBars :many +SELECT instrument_id, timeframe, timestamp, open, high, low, close, volume +FROM bars +WHERE instrument_id = $1 AND timeframe = $2 AND timestamp >= $3 AND timestamp <= $4 +ORDER BY timestamp ASC; + +-- name: UpsertRun :exec +INSERT INTO backtest_runs (id, strategy_id, market, timeframe, from_time, to_time, status, created_at, updated_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +ON CONFLICT (id) DO UPDATE SET + strategy_id = EXCLUDED.strategy_id, + market = EXCLUDED.market, + timeframe = EXCLUDED.timeframe, + from_time = EXCLUDED.from_time, + to_time = EXCLUDED.to_time, + status = EXCLUDED.status, + created_at = EXCLUDED.created_at, + updated_at = EXCLUDED.updated_at; + +-- name: GetRun :one +SELECT id, strategy_id, market, timeframe, from_time, to_time, status, created_at, updated_at +FROM backtest_runs +WHERE id = $1; diff --git a/services/worker/internal/storage/postgres/sqlc/db.go b/services/worker/internal/storage/postgres/sqlc/db.go new file mode 100644 index 0000000..b931bc5 --- /dev/null +++ b/services/worker/internal/storage/postgres/sqlc/db.go @@ -0,0 +1,32 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 + +package sqlc + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +type DBTX interface { + Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error) + Query(context.Context, string, ...interface{}) (pgx.Rows, error) + QueryRow(context.Context, string, ...interface{}) pgx.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +type Queries struct { + db DBTX +} + +func (q *Queries) WithTx(tx pgx.Tx) *Queries { + return &Queries{ + db: tx, + } +} diff --git a/services/worker/internal/storage/postgres/sqlc/models.go b/services/worker/internal/storage/postgres/sqlc/models.go new file mode 100644 index 0000000..25d18e7 --- /dev/null +++ b/services/worker/internal/storage/postgres/sqlc/models.go @@ -0,0 +1,42 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 + +package sqlc + +import ( + "github.com/jackc/pgx/v5/pgtype" +) + +type BacktestRun struct { + ID string + StrategyID string + Market string + Timeframe string + FromTime pgtype.Timestamptz + ToTime pgtype.Timestamptz + Status string + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz +} + +type Bar struct { + InstrumentID string + Timeframe string + Timestamp pgtype.Timestamptz + Open pgtype.Numeric + High pgtype.Numeric + Low pgtype.Numeric + Close pgtype.Numeric + Volume pgtype.Numeric +} + +type Instrument struct { + ID string + Market string + Venue string + Symbol string + Name string + Currency string + ProviderSymbols []byte +} diff --git a/services/worker/internal/storage/postgres/sqlc/queries.sql.go b/services/worker/internal/storage/postgres/sqlc/queries.sql.go new file mode 100644 index 0000000..a549009 --- /dev/null +++ b/services/worker/internal/storage/postgres/sqlc/queries.sql.go @@ -0,0 +1,249 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 +// source: queries.sql + +package sqlc + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const getBars = `-- name: GetBars :many +SELECT instrument_id, timeframe, timestamp, open, high, low, close, volume +FROM bars +WHERE instrument_id = $1 AND timeframe = $2 AND timestamp >= $3 AND timestamp <= $4 +ORDER BY timestamp ASC +` + +type GetBarsParams struct { + InstrumentID string + Timeframe string + Timestamp pgtype.Timestamptz + Timestamp_2 pgtype.Timestamptz +} + +func (q *Queries) GetBars(ctx context.Context, arg GetBarsParams) ([]Bar, error) { + rows, err := q.db.Query(ctx, getBars, + arg.InstrumentID, + arg.Timeframe, + arg.Timestamp, + arg.Timestamp_2, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Bar + for rows.Next() { + var i Bar + if err := rows.Scan( + &i.InstrumentID, + &i.Timeframe, + &i.Timestamp, + &i.Open, + &i.High, + &i.Low, + &i.Close, + &i.Volume, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getInstrument = `-- name: GetInstrument :one +SELECT id, market, venue, symbol, name, currency, provider_symbols +FROM instruments +WHERE id = $1 +` + +func (q *Queries) GetInstrument(ctx context.Context, id string) (Instrument, error) { + row := q.db.QueryRow(ctx, getInstrument, id) + var i Instrument + err := row.Scan( + &i.ID, + &i.Market, + &i.Venue, + &i.Symbol, + &i.Name, + &i.Currency, + &i.ProviderSymbols, + ) + return i, err +} + +const getRun = `-- name: GetRun :one +SELECT id, strategy_id, market, timeframe, from_time, to_time, status, created_at, updated_at +FROM backtest_runs +WHERE id = $1 +` + +func (q *Queries) GetRun(ctx context.Context, id string) (BacktestRun, error) { + row := q.db.QueryRow(ctx, getRun, id) + var i BacktestRun + err := row.Scan( + &i.ID, + &i.StrategyID, + &i.Market, + &i.Timeframe, + &i.FromTime, + &i.ToTime, + &i.Status, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const listInstruments = `-- name: ListInstruments :many +SELECT id, market, venue, symbol, name, currency, provider_symbols +FROM instruments +` + +func (q *Queries) ListInstruments(ctx context.Context) ([]Instrument, error) { + rows, err := q.db.Query(ctx, listInstruments) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Instrument + for rows.Next() { + var i Instrument + if err := rows.Scan( + &i.ID, + &i.Market, + &i.Venue, + &i.Symbol, + &i.Name, + &i.Currency, + &i.ProviderSymbols, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const upsertBar = `-- name: UpsertBar :exec +INSERT INTO bars (instrument_id, timeframe, timestamp, open, high, low, close, volume) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +ON CONFLICT (instrument_id, timeframe, timestamp) DO UPDATE SET + open = EXCLUDED.open, + high = EXCLUDED.high, + low = EXCLUDED.low, + close = EXCLUDED.close, + volume = EXCLUDED.volume +` + +type UpsertBarParams struct { + InstrumentID string + Timeframe string + Timestamp pgtype.Timestamptz + Open pgtype.Numeric + High pgtype.Numeric + Low pgtype.Numeric + Close pgtype.Numeric + Volume pgtype.Numeric +} + +func (q *Queries) UpsertBar(ctx context.Context, arg UpsertBarParams) error { + _, err := q.db.Exec(ctx, upsertBar, + arg.InstrumentID, + arg.Timeframe, + arg.Timestamp, + arg.Open, + arg.High, + arg.Low, + arg.Close, + arg.Volume, + ) + return err +} + +const upsertInstrument = `-- name: UpsertInstrument :exec +INSERT INTO instruments (id, market, venue, symbol, name, currency, provider_symbols) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (id) DO UPDATE SET + market = EXCLUDED.market, + venue = EXCLUDED.venue, + symbol = EXCLUDED.symbol, + name = EXCLUDED.name, + currency = EXCLUDED.currency, + provider_symbols = EXCLUDED.provider_symbols +` + +type UpsertInstrumentParams struct { + ID string + Market string + Venue string + Symbol string + Name string + Currency string + ProviderSymbols []byte +} + +func (q *Queries) UpsertInstrument(ctx context.Context, arg UpsertInstrumentParams) error { + _, err := q.db.Exec(ctx, upsertInstrument, + arg.ID, + arg.Market, + arg.Venue, + arg.Symbol, + arg.Name, + arg.Currency, + arg.ProviderSymbols, + ) + return err +} + +const upsertRun = `-- name: UpsertRun :exec +INSERT INTO backtest_runs (id, strategy_id, market, timeframe, from_time, to_time, status, created_at, updated_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +ON CONFLICT (id) DO UPDATE SET + strategy_id = EXCLUDED.strategy_id, + market = EXCLUDED.market, + timeframe = EXCLUDED.timeframe, + from_time = EXCLUDED.from_time, + to_time = EXCLUDED.to_time, + status = EXCLUDED.status, + created_at = EXCLUDED.created_at, + updated_at = EXCLUDED.updated_at +` + +type UpsertRunParams struct { + ID string + StrategyID string + Market string + Timeframe string + FromTime pgtype.Timestamptz + ToTime pgtype.Timestamptz + Status string + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz +} + +func (q *Queries) UpsertRun(ctx context.Context, arg UpsertRunParams) error { + _, err := q.db.Exec(ctx, upsertRun, + arg.ID, + arg.StrategyID, + arg.Market, + arg.Timeframe, + arg.FromTime, + arg.ToTime, + arg.Status, + arg.CreatedAt, + arg.UpdatedAt, + ) + return err +} diff --git a/services/worker/internal/storage/postgres/store.go b/services/worker/internal/storage/postgres/store.go new file mode 100644 index 0000000..f864b72 --- /dev/null +++ b/services/worker/internal/storage/postgres/store.go @@ -0,0 +1,133 @@ +package postgres + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.toki-labs.com/toki/alt/packages/domain/backtest" + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/storage" + "git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres/sqlc" +) + +var _ storage.InstrumentStore = (*Store)(nil) +var _ storage.BarStore = (*Store)(nil) +var _ storage.BacktestRunStore = (*Store)(nil) + +type Store struct { + pool *pgxpool.Pool + queries *sqlc.Queries +} + +func NewStore(pool *pgxpool.Pool) *Store { + return &Store{ + pool: pool, + queries: sqlc.New(pool), + } +} + +func (s *Store) UpsertInstrument(ctx context.Context, inst market.Instrument) error { + params, err := mapInstrumentToParams(inst) + if err != nil { + return fmt.Errorf("failed to map instrument: %w", err) + } + if err := s.queries.UpsertInstrument(ctx, params); err != nil { + return fmt.Errorf("failed to upsert instrument: %w", err) + } + return nil +} + +func (s *Store) GetInstrument(ctx context.Context, id market.InstrumentID) (market.Instrument, error) { + row, err := s.queries.GetInstrument(ctx, string(id)) + if err != nil { + return market.Instrument{}, fmt.Errorf("failed to get instrument: %w", err) + } + inst, err := mapRowToInstrument(row) + if err != nil { + return market.Instrument{}, fmt.Errorf("failed to map instrument row: %w", err) + } + return inst, nil +} + +func (s *Store) ListInstruments(ctx context.Context) ([]market.Instrument, error) { + rows, err := s.queries.ListInstruments(ctx) + if err != nil { + return nil, fmt.Errorf("failed to list instruments: %w", err) + } + insts := make([]market.Instrument, len(rows)) + for i, row := range rows { + inst, err := mapRowToInstrument(row) + if err != nil { + return nil, fmt.Errorf("failed to map instrument row at index %d: %w", i, err) + } + insts[i] = inst + } + return insts, nil +} + +func (s *Store) UpsertBar(ctx context.Context, bar market.Bar) error { + params, err := mapBarToParams(bar) + if err != nil { + return fmt.Errorf("failed to map bar: %w", err) + } + if err := s.queries.UpsertBar(ctx, params); err != nil { + return fmt.Errorf("failed to upsert bar: %w", err) + } + return nil +} + +func (s *Store) GetBars(ctx context.Context, id market.InstrumentID, timeframe market.Timeframe, from, to time.Time) ([]market.Bar, error) { + // First, fetch the instrument to know its currency + instRow, err := s.queries.GetInstrument(ctx, string(id)) + if err != nil { + return nil, fmt.Errorf("failed to get instrument for bar currency: %w", err) + } + + params := sqlc.GetBarsParams{ + InstrumentID: string(id), + Timeframe: string(timeframe), + Timestamp: pgtype.Timestamptz{Time: from, Valid: !from.IsZero()}, + Timestamp_2: pgtype.Timestamptz{Time: to, Valid: !to.IsZero()}, + } + rows, err := s.queries.GetBars(ctx, params) + if err != nil { + return nil, fmt.Errorf("failed to get bars: %w", err) + } + + bars := make([]market.Bar, len(rows)) + for i, row := range rows { + bar, err := mapRowToBar(row, market.Currency(instRow.Currency)) + if err != nil { + return nil, fmt.Errorf("failed to map bar row at index %d: %w", i, err) + } + bars[i] = bar + } + return bars, nil +} + +func (s *Store) UpsertRun(ctx context.Context, run backtest.Run) error { + params, err := mapRunToParams(run) + if err != nil { + return fmt.Errorf("failed to map run: %w", err) + } + if err := s.queries.UpsertRun(ctx, params); err != nil { + return fmt.Errorf("failed to upsert run: %w", err) + } + return nil +} + +func (s *Store) GetRun(ctx context.Context, id backtest.RunID) (backtest.Run, error) { + row, err := s.queries.GetRun(ctx, string(id)) + if err != nil { + return backtest.Run{}, fmt.Errorf("failed to get run: %w", err) + } + run, err := mapRowToRun(row) + if err != nil { + return backtest.Run{}, fmt.Errorf("failed to map run row: %w", err) + } + return run, nil +} diff --git a/services/worker/sqlc.yaml b/services/worker/sqlc.yaml new file mode 100644 index 0000000..b8e7c54 --- /dev/null +++ b/services/worker/sqlc.yaml @@ -0,0 +1,10 @@ +version: "2" +sql: + - engine: "postgresql" + schema: "internal/storage/postgres/migrations" + queries: "internal/storage/postgres/queries" + gen: + go: + package: "sqlc" + out: "internal/storage/postgres/sqlc" + sql_package: "pgx/v5" diff --git a/services/worker/tools.go b/services/worker/tools.go new file mode 100644 index 0000000..693ef44 --- /dev/null +++ b/services/worker/tools.go @@ -0,0 +1,8 @@ +//go:build tools +// +build tools + +package tools + +import ( + _ "github.com/sqlc-dev/sqlc/cmd/sqlc" +)