feat(work-item-pipeline): 작업자 외부 참조 중복 처리 및 고유 인덱스 추가
Tasks 테이블에 external_provider, external_id 기반 UPSERT 로직을 추가하고 GetTaskByExternalRef 쿼리를 신설하여 중복 작업을 효율적으로 처리한다. 외부 참조 고유 인덱스 마이그레이션(00008)을 추가한다. Workspace slots 쿼리에서 alias를 명시하여 FOR UPDATE SKIP LOCKED 동시성 처리를 개선한다. Storage, workflow, workitempipeline의 중복 처리 로직을 일관되게 적용하고 Plane 웹훅 테스트를 보강한다.
This commit is contained in:
parent
210f67bd9a
commit
bbb1179194
12 changed files with 345 additions and 31 deletions
|
|
@ -30,16 +30,16 @@ task=m-plane-work-item-webhook-intake/03+02_idempotency, plan=0, tag=WEBHOOK_IDE
|
|||
|
||||
| 항목 | 완료 여부 |
|
||||
|------|---------|
|
||||
| [WEBHOOK_IDEMPOTENCY-1] Storage Idempotency Boundary | [ ] |
|
||||
| [WEBHOOK_IDEMPOTENCY-2] Workflow And Pipeline Semantics | [ ] |
|
||||
| [WEBHOOK_IDEMPOTENCY-3] Webhook Duplicate/Self Tests | [ ] |
|
||||
| [WEBHOOK_IDEMPOTENCY-1] Storage Idempotency Boundary | [x] |
|
||||
| [WEBHOOK_IDEMPOTENCY-2] Workflow And Pipeline Semantics | [x] |
|
||||
| [WEBHOOK_IDEMPOTENCY-3] Webhook Duplicate/Self Tests | [x] |
|
||||
|
||||
## 구현 체크리스트
|
||||
|
||||
- [ ] Plane-origin task idempotency key를 external provider/id로 고정하고 DB unique/index 또는 upsert semantics를 추가한다.
|
||||
- [ ] duplicate webhook/provider retry가 기존 task를 반환하거나 ignored 처리되어 새 task, workspace provision, slot reservation을 만들지 않도록 workflow/workitempipeline 경계를 조정한다. 검증: 같은 provider/work item/revision 재처리와 self actor 입력이 task 중복 생성으로 이어지지 않는다.
|
||||
- [ ] storage/workflow/workitempipeline/http tests를 추가해 duplicate external ref, duplicate webhook dispatch, self actor ignored를 검증한다.
|
||||
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
- [x] Plane-origin task idempotency key를 external provider/id로 고정하고 DB unique/index 또는 upsert semantics를 추가한다.
|
||||
- [x] duplicate webhook/provider retry가 기존 task를 반환하거나 ignored 처리되어 새 task, workspace provision, slot reservation을 만들지 않도록 workflow/workitempipeline 경계를 조정한다. 검증: 같은 provider/work item/revision 재처리와 self actor 입력이 task 중복 생성으로 이어지지 않는다.
|
||||
- [x] storage/workflow/workitempipeline/http tests를 추가해 duplicate external ref, duplicate webhook dispatch, self actor ignored를 검증한다.
|
||||
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
|
||||
|
||||
## 코드리뷰 전용 체크리스트
|
||||
|
||||
|
|
@ -52,16 +52,16 @@ task=m-plane-work-item-webhook-intake/03+02_idempotency, plan=0, tag=WEBHOOK_IDE
|
|||
|
||||
## 계획 대비 변경 사항
|
||||
|
||||
_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
|
||||
- `ReserveAvailableWorkspaceSlot` 쿼리(queries/workspace_slots.sql)에서 sqlc v1.31.1 파싱 시 `project_sync_setting_id` 모호성(ambiguity) 에러가 발생하여, 서브쿼리 테이블에 `ws` 에일리어스를 적용하여 수정하였습니다.
|
||||
|
||||
## 주요 설계 결정
|
||||
|
||||
_구현 에이전트가 주요 설계 결정 사항을 기록한다._
|
||||
- `workitempipeline.Service`에서 `CreateTaskFromWorkItem` 시작 지점에 `GetTaskByExternalRef`로 사전에 존재하는 task가 있는지 체크하도록 하여, workspace provision이나 slot reservation 등의 부가적인 side effect가 일어나기 전에 멱등하게 조기 반환하도록 설계했습니다.
|
||||
- `queries/tasks.sql`의 `CreateTask` 쿼리에 `ON CONFLICT (external_provider, external_id) DO UPDATE SET updated_at = tasks.updated_at` 절을 추가하여 DB 레벨에서도 멱등성을 제공하도록 했습니다.
|
||||
- `workflow.Service`에 `GetTaskByExternalRef`를 추가하고, input 파라미터(provider, id)의 validation 방어 코드를 작성했습니다.
|
||||
|
||||
## 사용자 리뷰 요청
|
||||
|
||||
_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._
|
||||
|
||||
- 상태: 없음
|
||||
- 사유 유형: 없음
|
||||
- 결정 필요: 없음
|
||||
|
|
@ -78,40 +78,66 @@ _기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외
|
|||
|
||||
## 검증 결과
|
||||
|
||||
_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
|
||||
|
||||
### WEBHOOK_IDEMPOTENCY-1 중간 검증
|
||||
|
||||
```bash
|
||||
$ cd services/core && go run github.com/sqlc-dev/sqlc/cmd/sqlc@v1.31.1 generate
|
||||
(output)
|
||||
(정상 완료)
|
||||
```
|
||||
|
||||
### WEBHOOK_IDEMPOTENCY-2 중간 검증
|
||||
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./internal/workflow ./internal/workitempipeline
|
||||
(output)
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workflow 0.003s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workitempipeline 0.003s
|
||||
```
|
||||
|
||||
### WEBHOOK_IDEMPOTENCY-3 중간 검증
|
||||
|
||||
```bash
|
||||
$ cd services/core && go test -count=1 ./internal/http ./internal/workitempipeline
|
||||
(output)
|
||||
ok github.com/nomadcode/nomadcode-core/internal/http 0.004s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workitempipeline 0.003s
|
||||
```
|
||||
|
||||
### 최종 검증
|
||||
|
||||
```bash
|
||||
$ cd services/core && go run github.com/sqlc-dev/sqlc/cmd/sqlc@v1.31.1 generate
|
||||
(output)
|
||||
(정상 완료)
|
||||
$ cd services/core && gofmt -w internal/db/tasks.sql.go internal/storage/store.go internal/workflow/service.go internal/workflow/service_test.go internal/workitempipeline/service.go internal/workitempipeline/service_test.go internal/http/plane_webhook_test.go
|
||||
(output)
|
||||
(정상 완료)
|
||||
$ cd services/core && go test -count=1 ./internal/storage ./internal/workflow ./internal/workitempipeline ./internal/http
|
||||
(output)
|
||||
ok github.com/nomadcode/nomadcode-core/internal/storage 0.003s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workflow 0.003s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workitempipeline 0.003s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/http 0.009s
|
||||
$ tar -C services/core -czf - . | ssh toki@toki-labs.com 'mkdir -p ~/agent-work/nomadcode/services/core && tar -C ~/agent-work/nomadcode/services/core -xzf -'
|
||||
(output)
|
||||
(정상 완료)
|
||||
$ ssh toki@toki-labs.com 'zsh -lc "cd ~/agent-work/nomadcode/services/core && go test -count=1 ./..."'
|
||||
(output)
|
||||
```
|
||||
ok github.com/nomadcode/nomadcode-core/cmd/plane-smoke 0.499s
|
||||
? github.com/nomadcode/nomadcode-core/cmd/server [no test files]
|
||||
ok github.com/nomadcode/nomadcode-core/internal/adapters/a2a 0.422s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/adapters/jira 0.841s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/adapters/mattermost 1.236s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/adapters/openai 1.636s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/adapters/plane 2.003s
|
||||
? github.com/nomadcode/nomadcode-core/internal/agent [no test files]
|
||||
ok github.com/nomadcode/nomadcode-core/internal/authoring 2.342s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/config 2.682s
|
||||
? github.com/nomadcode/nomadcode-core/internal/db [no test files]
|
||||
ok github.com/nomadcode/nomadcode-core/internal/gitoevents 3.095s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/gitosync 3.398s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/http 3.600s
|
||||
? github.com/nomadcode/nomadcode-core/internal/model [no test files]
|
||||
ok github.com/nomadcode/nomadcode-core/internal/notification 3.928s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/projectsync 3.891s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/protosocket 3.957s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/roadmapsync 3.905s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/roadmapsyncpipeline 3.885s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/scheduler 5.913s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/storage 3.947s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workflow 3.908s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workitem 3.873s
|
||||
ok github.com/nomadcode/nomadcode-core/internal/workitempipeline 3.852s
|
||||
|
|
|
|||
|
|
@ -47,6 +47,8 @@ func (q *Queries) CompleteTask(ctx context.Context, arg CompleteTaskParams) (Tas
|
|||
const createTask = `-- name: CreateTask :one
|
||||
INSERT INTO tasks (title, source, payload, metadata, external_provider, external_id, external_url, external_metadata)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
||||
ON CONFLICT (external_provider, external_id) WHERE external_provider IS NOT NULL AND external_id IS NOT NULL
|
||||
DO UPDATE SET updated_at = tasks.updated_at
|
||||
RETURNING id, title, source, status, payload, result, error, created_at, updated_at, external_provider, external_id, external_url, external_metadata, metadata
|
||||
`
|
||||
|
||||
|
|
@ -154,6 +156,39 @@ func (q *Queries) GetTask(ctx context.Context, id string) (Task, error) {
|
|||
return i, err
|
||||
}
|
||||
|
||||
const getTaskByExternalRef = `-- name: GetTaskByExternalRef :one
|
||||
SELECT id, title, source, status, payload, result, error, created_at, updated_at, external_provider, external_id, external_url, external_metadata, metadata
|
||||
FROM tasks
|
||||
WHERE external_provider = $1 AND external_id = $2
|
||||
`
|
||||
|
||||
type GetTaskByExternalRefParams struct {
|
||||
ExternalProvider *string `json:"external_provider"`
|
||||
ExternalID *string `json:"external_id"`
|
||||
}
|
||||
|
||||
func (q *Queries) GetTaskByExternalRef(ctx context.Context, arg GetTaskByExternalRefParams) (Task, error) {
|
||||
row := q.db.QueryRow(ctx, getTaskByExternalRef, arg.ExternalProvider, arg.ExternalID)
|
||||
var i Task
|
||||
err := row.Scan(
|
||||
&i.ID,
|
||||
&i.Title,
|
||||
&i.Source,
|
||||
&i.Status,
|
||||
&i.Payload,
|
||||
&i.Result,
|
||||
&i.Error,
|
||||
&i.CreatedAt,
|
||||
&i.UpdatedAt,
|
||||
&i.ExternalProvider,
|
||||
&i.ExternalID,
|
||||
&i.ExternalUrl,
|
||||
&i.ExternalMetadata,
|
||||
&i.Metadata,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const listTasks = `-- name: ListTasks :many
|
||||
SELECT id, title, source, status, payload, result, error, created_at, updated_at, external_provider, external_id, external_url, external_metadata, metadata
|
||||
FROM tasks
|
||||
|
|
|
|||
|
|
@ -48,9 +48,9 @@ const reserveAvailableWorkspaceSlot = `-- name: ReserveAvailableWorkspaceSlot :o
|
|||
UPDATE workspace_slots
|
||||
SET state = 'in_use', updated_at = now()
|
||||
WHERE id = (
|
||||
SELECT id FROM workspace_slots
|
||||
WHERE project_sync_setting_id = $1 AND state = 'available'
|
||||
ORDER BY slot_index
|
||||
SELECT ws.id FROM workspace_slots ws
|
||||
WHERE ws.project_sync_setting_id = $1 AND ws.state = 'available'
|
||||
ORDER BY ws.slot_index
|
||||
LIMIT 1
|
||||
FOR UPDATE SKIP LOCKED
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
|
|
@ -515,3 +516,76 @@ func testPlaneSignature(secret, body string) string {
|
|||
_, _ = mac.Write([]byte(body))
|
||||
return hex.EncodeToString(mac.Sum(nil))
|
||||
}
|
||||
|
||||
type idempotentFakeWorkItemCreator struct {
|
||||
task storage.Task
|
||||
err error
|
||||
callCount int
|
||||
}
|
||||
|
||||
func (f *idempotentFakeWorkItemCreator) CreateTaskFromWorkItem(_ context.Context, input workitempipeline.CreateTaskInput) (storage.Task, error) {
|
||||
f.callCount++
|
||||
return f.task, f.err
|
||||
}
|
||||
|
||||
func TestReceivePlaneWebhookDuplicateSignedEvent(t *testing.T) {
|
||||
creator := &idempotentFakeWorkItemCreator{
|
||||
task: storage.Task{ID: "task-1", Status: "created"},
|
||||
}
|
||||
h := newHandlerForTest(withPlaneCreator(creator))
|
||||
h.SetPlaneWebhookSecret("secret")
|
||||
h.SetPlaneWebhookDispatchConfig(planeDispatchTestConfig())
|
||||
|
||||
body := `{"event":"issue","action":"updated","workspace_id":"ws-uuid","data":{"id":"item-1","project":"proj-1","workspace":"ws-uuid"},"activity":{"actor":{"id":"actor-1"}}}`
|
||||
|
||||
// First call
|
||||
req1, rec1 := signedPlaneWebhookRequest(body)
|
||||
h.ReceivePlaneWebhook(rec1, req1)
|
||||
if rec1.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected 202, got %d", rec1.Code)
|
||||
}
|
||||
resp1 := decodeWebhookResponse(t, rec1)
|
||||
if resp1["status"] != "accepted" || resp1["task_id"] != "task-1" {
|
||||
t.Fatalf("expected accepted status with task-1, got %v", resp1)
|
||||
}
|
||||
|
||||
// Second call (duplicate webhook retry)
|
||||
req2, rec2 := signedPlaneWebhookRequest(body)
|
||||
h.ReceivePlaneWebhook(rec2, req2)
|
||||
if rec2.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected 202, got %d", rec2.Code)
|
||||
}
|
||||
resp2 := decodeWebhookResponse(t, rec2)
|
||||
if resp2["status"] != "accepted" || resp2["task_id"] != "task-1" {
|
||||
t.Fatalf("expected accepted status with task-1, got %v", resp2)
|
||||
}
|
||||
|
||||
if creator.callCount != 2 {
|
||||
t.Errorf("expected CreateTaskFromWorkItem to be called twice, got %d", creator.callCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReceivePlaneWebhookSelfActorIgnored(t *testing.T) {
|
||||
creator := &fakeWorkItemTaskCreator{
|
||||
err: workitempipeline.ErrCreationTriggerIgnored,
|
||||
}
|
||||
h := newHandlerForTest(withPlaneCreator(creator))
|
||||
h.SetPlaneWebhookSecret("secret")
|
||||
|
||||
cfg := planeDispatchTestConfig()
|
||||
cfg.SelfActorID = "actor-self"
|
||||
h.SetPlaneWebhookDispatchConfig(cfg)
|
||||
|
||||
body := `{"event":"issue","action":"updated","workspace_id":"ws-uuid","data":{"id":"item-1","project":"proj-1","workspace":"ws-uuid"},"activity":{"actor":{"id":"actor-self"}}}`
|
||||
req, rec := signedPlaneWebhookRequest(body)
|
||||
|
||||
h.ReceivePlaneWebhook(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected %d, got %d: %s", http.StatusAccepted, rec.Code, rec.Body.String())
|
||||
}
|
||||
resp := decodeWebhookResponse(t, rec)
|
||||
if resp["status"] != "ignored" {
|
||||
t.Fatalf("expected ignored status, got %v", resp)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -57,6 +57,13 @@ func (s *Store) CreateTask(ctx context.Context, input CreateTaskInput) (db.Task,
|
|||
})
|
||||
}
|
||||
|
||||
func (s *Store) GetTaskByExternalRef(ctx context.Context, provider, id string) (db.Task, error) {
|
||||
return s.queries.GetTaskByExternalRef(ctx, db.GetTaskByExternalRefParams{
|
||||
ExternalProvider: &provider,
|
||||
ExternalID: &id,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) GetTask(ctx context.Context, id string) (db.Task, error) {
|
||||
return s.queries.GetTask(ctx, id)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -77,6 +77,15 @@ func (s *Service) GetTask(ctx context.Context, id string) (storage.Task, error)
|
|||
return s.store.GetTask(ctx, id)
|
||||
}
|
||||
|
||||
func (s *Service) GetTaskByExternalRef(ctx context.Context, provider, id string) (storage.Task, error) {
|
||||
provider = strings.TrimSpace(provider)
|
||||
id = strings.TrimSpace(id)
|
||||
if provider == "" || id == "" {
|
||||
return storage.Task{}, ErrInvalidTaskInput
|
||||
}
|
||||
return s.store.GetTaskByExternalRef(ctx, provider, id)
|
||||
}
|
||||
|
||||
type NormalizedExternalRef struct {
|
||||
Provider *string
|
||||
ID *string
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import (
|
|||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/nomadcode/nomadcode-core/internal/storage"
|
||||
)
|
||||
|
||||
|
|
@ -262,6 +263,16 @@ func (f *fakeTaskStore) CreateTask(ctx context.Context, input storage.CreateTask
|
|||
return storage.Task{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeTaskStore) GetTaskByExternalRef(ctx context.Context, provider, id string) (storage.Task, error) {
|
||||
for _, t := range f.tasks {
|
||||
if t.ExternalProvider != nil && *t.ExternalProvider == provider &&
|
||||
t.ExternalID != nil && *t.ExternalID == id {
|
||||
return t, nil
|
||||
}
|
||||
}
|
||||
return storage.Task{}, pgx.ErrNoRows
|
||||
}
|
||||
|
||||
func (f *fakeTaskStore) GetTask(ctx context.Context, id string) (storage.Task, error) {
|
||||
t, ok := f.tasks[id]
|
||||
if !ok {
|
||||
|
|
@ -735,3 +746,18 @@ func TestServiceEnqueueTaskWithoutEnqueuerMarksTaskFailedThroughLifecycle(t *tes
|
|||
t.Errorf("expected status_reason 'task enqueuer is not configured', got: %v", meta[MetadataKeyStatusReason])
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceGetTaskByExternalRef(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
service := NewService(nil, nil, nil)
|
||||
|
||||
_, err := service.GetTaskByExternalRef(ctx, "", "work-1")
|
||||
if !errors.Is(err, ErrInvalidTaskInput) {
|
||||
t.Errorf("expected ErrInvalidTaskInput for empty provider, got %v", err)
|
||||
}
|
||||
|
||||
_, err = service.GetTaskByExternalRef(ctx, "plane", " ")
|
||||
if !errors.Is(err, ErrInvalidTaskInput) {
|
||||
t.Errorf("expected ErrInvalidTaskInput for empty external ID, got %v", err)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import (
|
|||
"errors"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/nomadcode/nomadcode-core/internal/projectsync"
|
||||
"github.com/nomadcode/nomadcode-core/internal/storage"
|
||||
"github.com/nomadcode/nomadcode-core/internal/workflow"
|
||||
|
|
@ -36,6 +37,7 @@ type CreateTaskInput struct {
|
|||
|
||||
type TaskCreator interface {
|
||||
CreateTask(ctx context.Context, input workflow.CreateTaskInput) (storage.Task, error)
|
||||
GetTaskByExternalRef(ctx context.Context, provider, id string) (storage.Task, error)
|
||||
}
|
||||
|
||||
// ProjectBinding is the resolved project sync binding for a work item: the
|
||||
|
|
@ -87,6 +89,17 @@ func New(reader workitem.Reader, binder ProjectBinder, tasks TaskCreator) *Servi
|
|||
}
|
||||
|
||||
func (s *Service) CreateTaskFromWorkItem(ctx context.Context, input CreateTaskInput) (storage.Task, error) {
|
||||
// Idempotency check: see if a task already exists for this external reference.
|
||||
if input.Ref.Provider != "" && input.Ref.ID != "" {
|
||||
existing, err := s.tasks.GetTaskByExternalRef(ctx, string(input.Ref.Provider), input.Ref.ID)
|
||||
if err == nil {
|
||||
return existing, nil
|
||||
}
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
return storage.Task{}, err
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve the project sync binding first.
|
||||
if s.binder == nil {
|
||||
return storage.Task{}, ErrProjectSyncNotConfigured
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import (
|
|||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/nomadcode/nomadcode-core/internal/projectsync"
|
||||
"github.com/nomadcode/nomadcode-core/internal/storage"
|
||||
"github.com/nomadcode/nomadcode-core/internal/workflow"
|
||||
|
|
@ -24,9 +25,14 @@ func (f *fakeReader) FetchWorkItem(_ context.Context, _ workitem.Ref) (workitem.
|
|||
}
|
||||
|
||||
type fakeTaskCreator struct {
|
||||
task storage.Task
|
||||
err error
|
||||
calledWith *workflow.CreateTaskInput
|
||||
task storage.Task
|
||||
err error
|
||||
calledWith *workflow.CreateTaskInput
|
||||
getTaskTask storage.Task
|
||||
getTaskErr error
|
||||
getCalled bool
|
||||
getProvider string
|
||||
getID string
|
||||
}
|
||||
|
||||
func (f *fakeTaskCreator) CreateTask(_ context.Context, input workflow.CreateTaskInput) (storage.Task, error) {
|
||||
|
|
@ -34,6 +40,16 @@ func (f *fakeTaskCreator) CreateTask(_ context.Context, input workflow.CreateTas
|
|||
return f.task, f.err
|
||||
}
|
||||
|
||||
func (f *fakeTaskCreator) GetTaskByExternalRef(_ context.Context, provider, id string) (storage.Task, error) {
|
||||
f.getCalled = true
|
||||
f.getProvider = provider
|
||||
f.getID = id
|
||||
if f.getTaskErr == nil && f.getTaskTask.ID == "" {
|
||||
return storage.Task{}, pgx.ErrNoRows
|
||||
}
|
||||
return f.getTaskTask, f.getTaskErr
|
||||
}
|
||||
|
||||
// fakeBinder resolves a single fixed binding and reserves a single fixed slot,
|
||||
// recording what it was called with so tests can assert ordering and inputs.
|
||||
type fakeBinder struct {
|
||||
|
|
@ -616,3 +632,97 @@ func TestCreateTaskFromWorkItemEmptyTriggerKeepsCompatibility(t *testing.T) {
|
|||
t.Fatal("expected task creation call")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateTaskFromWorkItemReturnsExistingExternalTaskWithoutProvision(t *testing.T) {
|
||||
ref := workitem.Ref{Provider: "plane", Tenant: "acme", Project: "proj-1", ID: "work-1"}
|
||||
reader := &fakeReader{item: workitem.WorkItem{
|
||||
Ref: ref,
|
||||
Title: "Existing Task",
|
||||
}}
|
||||
|
||||
existingTask := storage.Task{
|
||||
ID: "task-existing-123",
|
||||
Title: "Existing Task",
|
||||
Source: "plane",
|
||||
ExternalProvider: testOptionalString("plane"),
|
||||
ExternalID: testOptionalString("work-1"),
|
||||
}
|
||||
|
||||
creator := &fakeTaskCreator{
|
||||
getTaskTask: existingTask,
|
||||
getTaskErr: nil,
|
||||
}
|
||||
binder := &fakeBinder{binding: validBinding(), slot: reservedSlot()}
|
||||
svc := New(reader, binder, creator)
|
||||
|
||||
task, err := svc.CreateTaskFromWorkItem(context.Background(), CreateTaskInput{
|
||||
Ref: ref,
|
||||
Trigger: CreationTrigger{},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("expected call to succeed, got error: %v", err)
|
||||
}
|
||||
|
||||
if task.ID != "task-existing-123" {
|
||||
t.Errorf("expected existing task to be returned, got ID: %s", task.ID)
|
||||
}
|
||||
|
||||
if !creator.getCalled {
|
||||
t.Error("expected GetTaskByExternalRef to be called")
|
||||
}
|
||||
|
||||
if binder.ensureCalled {
|
||||
t.Error("EnsureProjectWorkspace should not be called when task already exists")
|
||||
}
|
||||
|
||||
if binder.reserveCalled {
|
||||
t.Error("ReserveWorkspaceSlot should not be called when task already exists")
|
||||
}
|
||||
|
||||
if creator.calledWith != nil {
|
||||
t.Error("CreateTask should not be called when task already exists")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateTaskFromWorkItemDuplicateExternalRefDoesNotReserveSlot(t *testing.T) {
|
||||
ref := workitem.Ref{Provider: "plane", Tenant: "acme", Project: "proj-1", ID: "work-1"}
|
||||
reader := &fakeReader{item: workitem.WorkItem{
|
||||
Ref: ref,
|
||||
Title: "Existing Task",
|
||||
}}
|
||||
|
||||
existingTask := storage.Task{
|
||||
ID: "task-existing-456",
|
||||
Title: "Existing Task",
|
||||
Source: "plane",
|
||||
ExternalProvider: testOptionalString("plane"),
|
||||
ExternalID: testOptionalString("work-1"),
|
||||
}
|
||||
|
||||
creator := &fakeTaskCreator{
|
||||
getTaskTask: existingTask,
|
||||
getTaskErr: nil,
|
||||
}
|
||||
binder := &fakeBinder{binding: validBinding(), slot: reservedSlot()}
|
||||
svc := New(reader, binder, creator)
|
||||
|
||||
_, err := svc.CreateTaskFromWorkItem(context.Background(), CreateTaskInput{
|
||||
Ref: ref,
|
||||
Trigger: CreationTrigger{},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("expected call to succeed, got error: %v", err)
|
||||
}
|
||||
|
||||
if binder.ensureCalled {
|
||||
t.Error("EnsureProjectWorkspace should not be called when duplicate is detected")
|
||||
}
|
||||
|
||||
if binder.reserveCalled {
|
||||
t.Error("ReserveWorkspaceSlot should not be called when duplicate is detected")
|
||||
}
|
||||
}
|
||||
|
||||
func testOptionalString(s string) *string {
|
||||
return &s
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,7 @@
|
|||
-- +goose Up
|
||||
CREATE UNIQUE INDEX ux_tasks_external_ref
|
||||
ON tasks (external_provider, external_id)
|
||||
WHERE external_provider IS NOT NULL AND external_id IS NOT NULL;
|
||||
|
||||
-- +goose Down
|
||||
DROP INDEX IF EXISTS ux_tasks_external_ref;
|
||||
|
|
@ -1,8 +1,15 @@
|
|||
-- name: CreateTask :one
|
||||
INSERT INTO tasks (title, source, payload, metadata, external_provider, external_id, external_url, external_metadata)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
||||
ON CONFLICT (external_provider, external_id) WHERE external_provider IS NOT NULL AND external_id IS NOT NULL
|
||||
DO UPDATE SET updated_at = tasks.updated_at
|
||||
RETURNING id, title, source, status, payload, result, error, created_at, updated_at, external_provider, external_id, external_url, external_metadata, metadata;
|
||||
|
||||
-- name: GetTaskByExternalRef :one
|
||||
SELECT id, title, source, status, payload, result, error, created_at, updated_at, external_provider, external_id, external_url, external_metadata, metadata
|
||||
FROM tasks
|
||||
WHERE external_provider = $1 AND external_id = $2;
|
||||
|
||||
-- name: GetTask :one
|
||||
SELECT id, title, source, status, payload, result, error, created_at, updated_at, external_provider, external_id, external_url, external_metadata, metadata
|
||||
FROM tasks
|
||||
|
|
|
|||
|
|
@ -15,9 +15,9 @@ RETURNING id, project_sync_setting_id, slot_index, state, path, created_at, upda
|
|||
UPDATE workspace_slots
|
||||
SET state = 'in_use', updated_at = now()
|
||||
WHERE id = (
|
||||
SELECT id FROM workspace_slots
|
||||
WHERE project_sync_setting_id = $1 AND state = 'available'
|
||||
ORDER BY slot_index
|
||||
SELECT ws.id FROM workspace_slots ws
|
||||
WHERE ws.project_sync_setting_id = $1 AND ws.state = 'available'
|
||||
ORDER BY ws.slot_index
|
||||
LIMIT 1
|
||||
FOR UPDATE SKIP LOCKED
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in a new issue