1
import { eq } from "drizzle-orm";
2
 
3
import { CommitSha, UnixTimestampMs } from "@/contracts";
4
import {
5
  D1SyncStatus,
6
  nullableTrusted,
7
  type FinalizeRunExecutionInput,
8
  type FinalizeRunExecutionResult,
9
  ProjectRunStatus,
10
  type ProjectRunTerminalStatus,
11
  type RecordRunResolvedCommitInput,
12
  type RequestRunCancelInput,
13
  expectTrusted,
14
} from "@/worker/contracts";
15
import * as projectSchema from "@/worker/db/durable/schema/project-do";
16
 
17
import { getRunRow } from "../repo";
18
import {
19
  ensureProjectState,
20
  getSnapshot,
21
  isTerminalStatus,
22
  nextMetadataD1SyncStatus,
23
  nextTerminalD1SyncStatus,
24
  promoteNextPendingRun,
25
} from "./shared";
26
import type { CancelTransitionResult, ProjectDoContext, ProjectRunRow, ProjectStore } from "../types";
27
 
28
const isFinalizeTransitionAllowed = (
29
  currentStatus: Extract<ProjectRunStatus, "active" | "cancel_requested">,
30
  terminalStatus: ProjectRunTerminalStatus,
31
): boolean =>
32
  currentStatus === "active"
33
    ? terminalStatus === "passed" || terminalStatus === "failed"
34
    : terminalStatus === "canceled" || terminalStatus === "failed";
35
 
36
export const transitionFinalizeRunExecution = (
37
  context: ProjectDoContext,
38
  tx: ProjectStore,
39
  input: FinalizeRunExecutionInput,
40
): FinalizeRunExecutionResult => {
41
  ensureProjectState(context, tx, input.projectId);
42
 
43
  const row = getRunRow(tx, input.projectId, input.runId);
44
  if (!row) {
45
    throw new Error(`Run ${input.runId} was not found in project state.`);
46
  }
47
 
48
  const currentStatus = expectTrusted(ProjectRunStatus, row.status, "ProjectRunStatus");
49
 
50
  if (currentStatus === "active" || currentStatus === "cancel_requested") {
51
    if (!isFinalizeTransitionAllowed(currentStatus, input.terminalStatus)) {
52
      throw new Error(`Run ${input.runId} cannot transition from ${currentStatus} to ${input.terminalStatus}.`);
53
    }
54
 
55
    tx.update(projectSchema.projectRuns)
56
      .set({
57
        status: input.terminalStatus,
58
        dispatchStatus: "terminal",
59
        d1SyncStatus: nextTerminalD1SyncStatus(expectTrusted(D1SyncStatus, row.d1SyncStatus, "D1SyncStatus")),
60
        lastError: input.lastError,
61
      })
62
      .where(eq(projectSchema.projectRuns.runId, input.runId))
63
      .run();
64
    tx.update(projectSchema.projectState)
65
      .set({
66
        activeRunId: null,
67
        updatedAt: Date.now(),
68
      })
69
      .where(eq(projectSchema.projectState.projectId, input.projectId))
70
      .run();
71
    promoteNextPendingRun(tx, input.projectId);
72
  } else if (isTerminalStatus(currentStatus)) {
73
    if (currentStatus !== input.terminalStatus) {
74
      throw new Error(
75
        `Run ${input.runId} is already terminal in status ${currentStatus}, cannot finalize as ${input.terminalStatus}.`,
76
      );
77
    }
78
  } else {
79
    throw new Error(`Run ${input.runId} cannot be finalized from status ${currentStatus}.`);
80
  }
81
 
82
  return {
83
    snapshot: getSnapshot(row),
84
  };
85
};
86
 
87
export type RecordRunResolvedCommitTransition =
88
  | {
89
      kind: "applied";
90
      row: ProjectRunRow;
91
    }
92
  | {
93
      kind: "stale";
94
      status: ProjectRunStatus | null;
95
    };
96
 
97
export const transitionRecordRunResolvedCommit = (
98
  context: ProjectDoContext,
99
  tx: ProjectStore,
100
  input: RecordRunResolvedCommitInput,
101
): RecordRunResolvedCommitTransition => {
102
  ensureProjectState(context, tx, input.projectId);
103
 
104
  const row = getRunRow(tx, input.projectId, input.runId);
105
  if (!row) {
106
    return {
107
      kind: "stale",
108
      status: null,
109
    };
110
  }
111
 
112
  const currentStatus = expectTrusted(ProjectRunStatus, row.status, "ProjectRunStatus");
113
  if (currentStatus !== "active" && currentStatus !== "cancel_requested") {
114
    return {
115
      kind: "stale",
116
      status: currentStatus,
117
    };
118
  }
119
 
120
  const currentCommitSha = nullableTrusted(CommitSha, row.commitSha, "CommitSha");
121
  if (currentCommitSha !== null && currentCommitSha !== input.commitSha) {
122
    throw new Error(`Run ${input.runId} already recorded commit ${currentCommitSha}, cannot replace it.`);
123
  }
124
 
125
  if (currentCommitSha === input.commitSha) {
126
    return {
127
      kind: "applied",
128
      row,
129
    };
130
  }
131
 
132
  const nextD1SyncStatus = nextMetadataD1SyncStatus(expectTrusted(D1SyncStatus, row.d1SyncStatus, "D1SyncStatus"));
133
  tx.update(projectSchema.projectRuns)
134
    .set({
135
      commitSha: input.commitSha,
136
      d1SyncStatus: nextD1SyncStatus,
137
    })
138
    .where(eq(projectSchema.projectRuns.runId, row.runId))
139
    .run();
140
 
141
  return {
142
    kind: "applied",
143
    row: {
144
      ...row,
145
      commitSha: input.commitSha,
146
      d1SyncStatus: nextD1SyncStatus,
147
    },
148
  };
149
};
150
 
151
export const transitionRequestRunCancel = (
152
  context: ProjectDoContext,
153
  tx: ProjectStore,
154
  input: RequestRunCancelInput,
155
  requestedAt: UnixTimestampMs,
156
): CancelTransitionResult => {
157
  ensureProjectState(context, tx, input.projectId);
158
 
159
  const row = getRunRow(tx, input.projectId, input.runId);
160
  if (!row) {
161
    throw new Error(`Run ${input.runId} was not found in project state.`);
162
  }
163
 
164
  if (row.status === "pending" || row.status === "executable") {
165
    tx.update(projectSchema.projectRuns)
166
      .set({
167
        status: "canceled",
168
        position: null,
169
        dispatchStatus: "terminal",
170
        d1SyncStatus: nextTerminalD1SyncStatus(expectTrusted(D1SyncStatus, row.d1SyncStatus, "D1SyncStatus")),
171
        cancelRequestedAt: requestedAt,
172
        lastError: null,
173
      })
174
      .where(eq(projectSchema.projectRuns.runId, row.runId))
175
      .run();
176
    promoteNextPendingRun(tx, input.projectId);
177
 
178
    return {
179
      row,
180
      runStatus: "canceled",
181
      cancelRequestedAt: requestedAt,
182
      runDoAction: "canceled",
183
    };
184
  }
185
 
186
  if (row.status === "active" || row.status === "cancel_requested") {
187
    const cancelRequestedAt = nullableTrusted(UnixTimestampMs, row.cancelRequestedAt, "UnixTimestampMs") ?? requestedAt;
188
    tx.update(projectSchema.projectRuns)
189
      .set({
190
        status: "cancel_requested",
191
        cancelRequestedAt,
192
      })
193
      .where(eq(projectSchema.projectRuns.runId, row.runId))
194
      .run();
195
 
196
    return {
197
      row,
198
      runStatus: "cancel_requested",
199
      cancelRequestedAt,
200
      runDoAction: "cancel_requested",
201
    };
202
  }
203
 
204
  return {
205
    row,
206
    runStatus: expectTrusted(ProjectRunStatus, row.status, "ProjectRunStatus"),
207
    cancelRequestedAt: nullableTrusted(UnixTimestampMs, row.cancelRequestedAt, "UnixTimestampMs"),
208
    runDoAction: "none",
209
  };
210
};