forked from molecule-ai/molecule-core
fix(org): DB-authoritative schedules; org/import is additive on template rows (#24)
Resolves #24 per CEO direction. DB is source of truth for workspace_schedules. POST /org/import becomes idempotent — only touches rows it owns (source='template'); runtime-added schedules (Canvas / API) are preserved across re-imports. - Migration 022: adds source TEXT NOT NULL DEFAULT 'runtime' CHECK in ('template','runtime'); unique index on (workspace_id, name) so the org/import upsert can use ON CONFLICT. - org.go: schedule INSERT becomes INSERT ... 'template' ON CONFLICT (workspace_id, name) DO UPDATE SET ... WHERE workspace_schedules.source='template'. Never DELETEs. - schedules.go: runtime POST writes 'runtime' explicitly; List handler surfaces the source field on the response so Canvas can render badges. - 3 new unit tests assert source='runtime' default for runtime CRUD, the SQL shape contract for org/import (additive + idempotent + runtime-preserving + never-DELETE), and List response surface. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hongming Wang
parent
dd61714c55
commit
2e9fb51ff9
@ -447,13 +447,28 @@ func (h *OrgHandler) createWorkspaceTree(ws OrgWorkspace, parentID *string, defa
|
||||
enabled = *sched.Enabled
|
||||
}
|
||||
nextRun, _ := scheduler.ComputeNextRun(sched.CronExpr, tz, time.Now())
|
||||
// Idempotent (additive) import. The DB is the source of truth (issue #24).
|
||||
// - INSERT new template-source rows when missing.
|
||||
// - On (workspace_id, name) collision, only refresh rows whose
|
||||
// source='template'. Runtime-added schedules with the same name
|
||||
// are preserved across re-imports.
|
||||
// - Never DELETE — rows present in DB but absent from the template
|
||||
// are left alone (could be runtime-added or removed-from-template).
|
||||
if _, err := db.DB.ExecContext(context.Background(), `
|
||||
INSERT INTO workspace_schedules (workspace_id, name, cron_expr, timezone, prompt, enabled, next_run_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
INSERT INTO workspace_schedules (workspace_id, name, cron_expr, timezone, prompt, enabled, next_run_at, source)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, 'template')
|
||||
ON CONFLICT (workspace_id, name) DO UPDATE
|
||||
SET cron_expr = EXCLUDED.cron_expr,
|
||||
timezone = EXCLUDED.timezone,
|
||||
prompt = EXCLUDED.prompt,
|
||||
enabled = EXCLUDED.enabled,
|
||||
next_run_at = EXCLUDED.next_run_at,
|
||||
updated_at = now()
|
||||
WHERE workspace_schedules.source = 'template'
|
||||
`, id, sched.Name, sched.CronExpr, tz, sched.Prompt, enabled, nextRun); err != nil {
|
||||
log.Printf("Org import: failed to create schedule '%s' for %s: %v", sched.Name, ws.Name, err)
|
||||
log.Printf("Org import: failed to upsert schedule '%s' for %s: %v", sched.Name, ws.Name, err)
|
||||
} else {
|
||||
log.Printf("Org import: schedule '%s' (%s) created for %s", sched.Name, sched.CronExpr, ws.Name)
|
||||
log.Printf("Org import: schedule '%s' (%s) upserted for %s (source=template)", sched.Name, sched.CronExpr, ws.Name)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -32,6 +32,7 @@ type scheduleResponse struct {
|
||||
RunCount int `json:"run_count"`
|
||||
LastStatus string `json:"last_status"`
|
||||
LastError string `json:"last_error"`
|
||||
Source string `json:"source"` // 'template' (seeded by org/import) | 'runtime' (created via Canvas/API). Issue #24.
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
@ -44,7 +45,7 @@ func (h *ScheduleHandler) List(c *gin.Context) {
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
SELECT id, workspace_id, name, cron_expr, timezone, prompt, enabled,
|
||||
last_run_at, next_run_at, run_count, last_status, last_error,
|
||||
created_at, updated_at
|
||||
source, created_at, updated_at
|
||||
FROM workspace_schedules
|
||||
WHERE workspace_id = $1
|
||||
ORDER BY created_at ASC
|
||||
@ -61,7 +62,7 @@ func (h *ScheduleHandler) List(c *gin.Context) {
|
||||
if err := rows.Scan(
|
||||
&s.ID, &s.WorkspaceID, &s.Name, &s.CronExpr, &s.Timezone,
|
||||
&s.Prompt, &s.Enabled, &s.LastRunAt, &s.NextRunAt, &s.RunCount,
|
||||
&s.LastStatus, &s.LastError, &s.CreatedAt, &s.UpdatedAt,
|
||||
&s.LastStatus, &s.LastError, &s.Source, &s.CreatedAt, &s.UpdatedAt,
|
||||
); err != nil {
|
||||
log.Printf("Schedules.List: scan error: %v", err)
|
||||
continue
|
||||
@ -117,9 +118,12 @@ func (h *ScheduleHandler) Create(c *gin.Context) {
|
||||
}
|
||||
|
||||
var id string
|
||||
// source='runtime' marks this row as user-created (Canvas/API). The
|
||||
// org/import path inserts with source='template' and only refreshes
|
||||
// template-source rows on re-import (issue #24), so runtime rows survive.
|
||||
err = db.DB.QueryRowContext(ctx, `
|
||||
INSERT INTO workspace_schedules (workspace_id, name, cron_expr, timezone, prompt, enabled, next_run_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
INSERT INTO workspace_schedules (workspace_id, name, cron_expr, timezone, prompt, enabled, next_run_at, source)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, 'runtime')
|
||||
RETURNING id
|
||||
`, workspaceID, body.Name, body.CronExpr, body.Timezone, body.Prompt, enabled, nextRun).Scan(&id)
|
||||
if err != nil {
|
||||
|
||||
148
platform/internal/handlers/schedules_test.go
Normal file
148
platform/internal/handlers/schedules_test.go
Normal file
@ -0,0 +1,148 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// Issue #24 — DB is the source of truth; org/import is additive on
|
||||
// template-source rows only. Runtime-added schedules survive re-imports.
|
||||
|
||||
// TestRuntimeSchedule_HasSourceRuntime asserts that POST /workspaces/:id/schedules
|
||||
// writes source='runtime' so that re-imports of the org template never touch
|
||||
// these user-created rows (preserved across re-imports).
|
||||
func TestRuntimeSchedule_HasSourceRuntime(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
setupTestRedis(t)
|
||||
handler := NewScheduleHandler()
|
||||
|
||||
// Match the literal 'runtime' source baked into the INSERT and capture
|
||||
// the workspace id arg. The inserted row id is returned via RETURNING.
|
||||
mock.ExpectQuery("INSERT INTO workspace_schedules .* VALUES .* 'runtime'").
|
||||
WithArgs("550e8400-e29b-41d4-a716-446655440000", "test", "*/5 * * * *", "UTC", "do thing", true, sqlmock.AnyArg()).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("11111111-1111-1111-1111-111111111111"))
|
||||
|
||||
body := []byte(`{"name":"test","cron_expr":"*/5 * * * *","prompt":"do thing"}`)
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: "550e8400-e29b-41d4-a716-446655440000"}}
|
||||
c.Request = httptest.NewRequest("POST", "/workspaces/550e8400-e29b-41d4-a716-446655440000/schedules", bytes.NewReader(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
handler.Create(c)
|
||||
|
||||
if w.Code != http.StatusCreated {
|
||||
t.Fatalf("expected 201, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatalf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestImport_OrgScheduleSQLShape verifies the SQL emitted by the org/import
|
||||
// path for schedules. It MUST be an INSERT ... ON CONFLICT (workspace_id, name)
|
||||
// DO UPDATE ... WHERE source='template' with VALUES ... 'template'. Together
|
||||
// these guarantee that re-import is:
|
||||
// - additive (new template rows are inserted),
|
||||
// - idempotent (existing template rows are refreshed),
|
||||
// - non-destructive of runtime rows (the WHERE filter skips them),
|
||||
// - never DELETE-based (additive only).
|
||||
//
|
||||
// This is a structural assertion against the source — cheap and catches a
|
||||
// regression that would silently break user-created schedules across
|
||||
// re-imports without needing a full provisioner harness.
|
||||
func TestImport_OrgScheduleSQLShape(t *testing.T) {
|
||||
got := orgImportScheduleSQL(t)
|
||||
|
||||
// Single test covers four CEO requirements at once: additive seed
|
||||
// (template marker), idempotent refresh (ON CONFLICT DO UPDATE),
|
||||
// runtime-row preservation (WHERE source='template'), and never-DELETE.
|
||||
mustContain := []string{
|
||||
"INSERT INTO workspace_schedules",
|
||||
"source",
|
||||
"'template'",
|
||||
"ON CONFLICT (workspace_id, name) DO UPDATE",
|
||||
"WHERE workspace_schedules.source = 'template'",
|
||||
}
|
||||
for _, s := range mustContain {
|
||||
if !strings.Contains(got, s) {
|
||||
t.Errorf("org/import schedule SQL missing fragment %q\n--- SQL ---\n%s", s, got)
|
||||
}
|
||||
}
|
||||
if regexp.MustCompile(`(?i)\bDELETE\b\s+FROM\s+workspace_schedules`).MatchString(got) {
|
||||
t.Error("org/import schedule SQL must never DELETE — additive only")
|
||||
}
|
||||
}
|
||||
|
||||
// orgImportScheduleSQL pulls the relevant SQL fragment from org.go for shape
|
||||
// assertions. Keeps this test independent of the provisioner.
|
||||
func orgImportScheduleSQL(t *testing.T) string {
|
||||
t.Helper()
|
||||
// Test runs from package dir (platform/internal/handlers) so org.go is local.
|
||||
data, err := os.ReadFile("org.go")
|
||||
if err != nil {
|
||||
t.Fatalf("read org.go: %v", err)
|
||||
}
|
||||
src := string(data)
|
||||
idx := strings.Index(src, "INSERT INTO workspace_schedules")
|
||||
if idx < 0 {
|
||||
t.Fatalf("could not find schedules INSERT in org.go")
|
||||
}
|
||||
end := strings.Index(src[idx:], "`")
|
||||
if end < 0 {
|
||||
t.Fatalf("could not find closing backtick after schedules INSERT")
|
||||
}
|
||||
return src[idx : idx+end]
|
||||
}
|
||||
|
||||
// TestList_IncludesSourceColumn asserts GET /workspaces/:id/schedules
|
||||
// returns the source field so Canvas can render template/runtime badges.
|
||||
func TestList_IncludesSourceColumn(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
setupTestRedis(t)
|
||||
handler := NewScheduleHandler()
|
||||
|
||||
cols := []string{
|
||||
"id", "workspace_id", "name", "cron_expr", "timezone", "prompt", "enabled",
|
||||
"last_run_at", "next_run_at", "run_count", "last_status", "last_error",
|
||||
"source", "created_at", "updated_at",
|
||||
}
|
||||
now := time.Now()
|
||||
mock.ExpectQuery("SELECT .* source, created_at, updated_at\\s+FROM workspace_schedules").
|
||||
WithArgs("550e8400-e29b-41d4-a716-446655440000").
|
||||
WillReturnRows(sqlmock.NewRows(cols).
|
||||
AddRow("id1", "550e8400-e29b-41d4-a716-446655440000", "tmpl-sched", "0 * * * *", "UTC", "p", true,
|
||||
nil, nil, 0, "", "", "template", now, now).
|
||||
AddRow("id2", "550e8400-e29b-41d4-a716-446655440000", "user-sched", "*/5 * * * *", "UTC", "p2", true,
|
||||
nil, nil, 0, "", "", "runtime", now, now))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Params = gin.Params{{Key: "id", Value: "550e8400-e29b-41d4-a716-446655440000"}}
|
||||
c.Request = httptest.NewRequest("GET", "/workspaces/550e8400-e29b-41d4-a716-446655440000/schedules", nil)
|
||||
|
||||
handler.List(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
body := w.Body.String()
|
||||
if !strings.Contains(body, `"source":"template"`) {
|
||||
t.Errorf(`response missing "source":"template": %s`, body)
|
||||
}
|
||||
if !strings.Contains(body, `"source":"runtime"`) {
|
||||
t.Errorf(`response missing "source":"runtime": %s`, body)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatalf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,2 @@
|
||||
DROP INDEX IF EXISTS idx_schedules_workspace_name;
|
||||
ALTER TABLE workspace_schedules DROP COLUMN IF EXISTS source;
|
||||
17
platform/migrations/022_workspace_schedules_source.up.sql
Normal file
17
platform/migrations/022_workspace_schedules_source.up.sql
Normal file
@ -0,0 +1,17 @@
|
||||
-- Add `source` column to workspace_schedules so the org-import path can
|
||||
-- distinguish rows it owns ('template') from rows created via the runtime
|
||||
-- API/Canvas ('runtime'). DB is the source of truth; org/import is now
|
||||
-- additive — it only INSERTs missing template rows and only UPDATEs rows
|
||||
-- where source = 'template'. Runtime-added schedules survive re-imports.
|
||||
|
||||
ALTER TABLE workspace_schedules
|
||||
ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'runtime'
|
||||
CHECK (source IN ('template', 'runtime'));
|
||||
|
||||
COMMENT ON COLUMN workspace_schedules.source IS
|
||||
'template = seeded by org/import (refreshable); runtime = created via Canvas/API (preserved across re-imports)';
|
||||
|
||||
-- Required so org-import can use ON CONFLICT (workspace_id, name) DO UPDATE.
|
||||
-- Schedules within a single workspace are uniquely identified by name.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_schedules_workspace_name
|
||||
ON workspace_schedules(workspace_id, name);
|
||||
Loading…
Reference in New Issue
Block a user