1import * as storagetransfer from "@distilled.cloud/gcp/storagetransfer_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { GcpEnvironment } from "../Environment.ts";
9import type { Providers } from "../Providers.ts";
10import {
11 createOwnership,
12 DEFAULT_JOB_STATUS,
13 encodeOwnership,
14 fieldMask,
15 getTransferJob,
16 hasOwnershipMarker,
17 isDeletedJob,
18 jobIdOf,
19 listTransferJobs,
20 ownedByAlchemy,
21 parseOwnership,
22 retryApiDisabled,
23 sameValue,
24 sanitizeJobId,
25 toJobId,
26 transferJobName,
27} from "./internal.ts";
28
29export type GcsData = storagetransfer.GcsData;
30export type PosixFilesystem = storagetransfer.PosixFilesystem;
31export type AwsS3Data = storagetransfer.AwsS3Data;
32export type HttpData = storagetransfer.HttpData;
33export type AzureBlobStorageData = storagetransfer.AzureBlobStorageData;
34export type AwsS3CompatibleData = storagetransfer.AwsS3CompatibleData;
35export type HdfsData = storagetransfer.HdfsData;
36export type ObjectConditions = storagetransfer.ObjectConditions;
37export type TransferOptions = storagetransfer.TransferOptions;
38export type TransferManifest = storagetransfer.TransferManifest;
39export type TransferSpec = storagetransfer.TransferSpec;
40export type ReplicationSpec = storagetransfer.ReplicationSpec;
41export type NotificationConfig = storagetransfer.NotificationConfig;
42export type LoggingConfig = storagetransfer.LoggingConfig;
43export type Schedule = storagetransfer.Schedule;
44export type EventStream = storagetransfer.EventStream;
45export type TransferJobStatus =
46 | storagetransfer.TransferJobStatusEnum
47 | (string & {});
48
49export type TransferJobProps = {
50 /**
51 * Job id (the `{job}` segment of `transferJobs/{job}`). If omitted, a
52 * unique name is generated from the stack, stage, and logical id. The
53 * API name is `transferJobs/{job}` (max 128 characters, must not start
54 * with `transferJobs/OPI` unless this is a POSIX transfer). Immutable
55 * — changing it replaces the job.
56 */
57 jobId?: string;
58 /**
59 * User description (max 1024 bytes). Transfer jobs have no labels
60 * field, so Alchemy ownership is stored in a `[alchemy …]` prefix and
61 * stripped from attributes.
62 */
63 description?: string;
64 /**
65 * Transfer specification (source, sink, filters, options). Required
66 * unless `replicationSpec` is set. Updates in place; a complete spec
67 * must be provided on each update.
68 */
69 transferSpec?: TransferSpec;
70 /**
71 * Cross-bucket replication specification. Mutually exclusive with
72 * `transferSpec`. Immutable after create — changing it replaces the
73 * job.
74 */
75 replicationSpec?: ReplicationSpec;
76 /**
77 * Recurring or one-time schedule. Optional; omit to create a job that
78 * never auto-runs (invoke `RunTransferJob` instead). Immutable after
79 * create — changing it replaces the job.
80 */
81 schedule?: Schedule;
82 /**
83 * Event-driven transfer stream. When set, `schedule` is ignored.
84 * Immutable after create — changing it replaces the job.
85 */
86 eventStream?: EventStream;
87 /**
88 * Pub/Sub notification configuration. Updates in place.
89 */
90 notificationConfig?: NotificationConfig;
91 /**
92 * Cloud Logging configuration. Updates in place.
93 */
94 loggingConfig?: LoggingConfig;
95 /**
96 * Job status. Must be `ENABLED` or `DISABLED` on create (the API
97 * rejects `STATUS_UNSPECIFIED`). `DELETED` is applied by delete.
98 * @default "ENABLED"
99 */
100 status?: TransferJobStatus;
101 /**
102 * User-managed service account that receives bucket permissions
103 * instead of the Transfer Service agent. Email or unique id. Immutable
104 * after create — changing it replaces the job.
105 */
106 serviceAccount?: string;
107};
108
109export type TransferJob = Resource<
110 "GCP.StorageTransfer.TransferJob",
111 TransferJobProps,
112 {
113 /** Full resource name `transferJobs/{job}`. */
114 name: string;
115 /** Job id (path segment after `transferJobs/`). */
116 jobId: string;
117 /** Project id. */
118 project: string;
119 /** User description with the Alchemy ownership prefix stripped. */
120 description: string | undefined;
121 /** Job status (`ENABLED`, `DISABLED`, `DELETED`). */
122 status: string | undefined;
123 /** Transfer specification. */
124 transferSpec: TransferSpec | undefined;
125 /** Replication specification. */
126 replicationSpec: ReplicationSpec | undefined;
127 /** Notification configuration. */
128 notificationConfig: NotificationConfig | undefined;
129 /** Logging configuration. */
130 loggingConfig: LoggingConfig | undefined;
131 /** Schedule, if any. */
132 schedule: Schedule | undefined;
133 /** Event stream, if any. */
134 eventStream: EventStream | undefined;
135 /** Delegated service account, if any. */
136 serviceAccount: string | undefined;
137 /** RFC3339 creation timestamp. */
138 creationTime: string | undefined;
139 /** RFC3339 last-modification timestamp. */
140 lastModificationTime: string | undefined;
141 /** Most recently started TransferOperation name, if any. */
142 latestOperationName: string | undefined;
143 },
144 never,
145 Providers
146>;
147
148/**
149 * A Storage Transfer Service job that copies objects between Cloud
150 * Storage, S3, Azure, POSIX, or HTTP sources and sinks.
151 *
152 * Transfer jobs have no labels field, so Alchemy stamps ownership into
153 * `description` for `list` / nuke. Name is identity. `description`,
154 * `transferSpec`, `notificationConfig`, `loggingConfig`, and `status`
155 * update in place. `schedule`, `eventStream`, `serviceAccount`, and
156 * `replicationSpec` are immutable after create.
157 *
158 * ### Creating a Transfer Job
159 * **Example:** One-shot GCS-to-GCS copy (disabled until RunTransferJob)
160 * ```typescript
161 * const job = yield* GCP.StorageTransfer.TransferJob("Nightly", {
162 * status: "DISABLED",
163 * description: "copy uploads into the archive bucket",
164 * transferSpec: {
165 * gcsDataSource: { bucketName: source.bucketName },
166 * gcsDataSink: { bucketName: archive.bucketName },
167 * },
168 * });
169 * ```
170 *
171 * **Example:** Recurring copy with a daily schedule
172 * ```typescript
173 * const job = yield* GCP.StorageTransfer.TransferJob("Nightly", {
174 * description: "daily archive",
175 * transferSpec: {
176 * gcsDataSource: { bucketName: "src-bucket", path: "uploads/" },
177 * gcsDataSink: { bucketName: "dst-bucket" },
178 * },
179 * schedule: {
180 * scheduleStartDate: { year: 2026, month: 1, day: 1 },
181 * startTimeOfDay: { hours: 3, minutes: 0 },
182 * repeatInterval: "86400s",
183 * },
184 * });
185 * ```
186 *
187 * ### Updating a Transfer Job
188 * Change props on the same logical id; the engine keeps the physical id.
189 *
190 * **Example:** Enable the job and change the source prefix
191 * ```typescript
192 * const job = yield* GCP.StorageTransfer.TransferJob("Nightly", {
193 * status: "ENABLED",
194 * description: "daily archive",
195 * transferSpec: {
196 * gcsDataSource: { bucketName: "src-bucket", path: "inbox/" },
197 * gcsDataSink: { bucketName: "dst-bucket" },
198 * },
199 * });
200 * ```
201 *
202 * @resource
203 * @category StorageTransfer
204 */
205export const TransferJob = Resource<TransferJob>(
206 "GCP.StorageTransfer.TransferJob",
207);
208
209export class TransferJobNotResolved extends Data.TaggedError(
210 "GCP.StorageTransfer.TransferJobNotResolved",
211)<{
212 name: string;
213}> {}
214
215const statusOf = (status: string | undefined) =>
216 status && status.length > 0 ? status : DEFAULT_JOB_STATUS;
217
218const toAttrs = (job: storagetransfer.TransferJob, project: string) => {
219 const name = job.name ?? "";
220 return {
221 name,
222 jobId: jobIdOf(name),
223 project: job.projectId ?? project,
224 description: parseOwnership(job.description).text,
225 status: job.status,
226 transferSpec: job.transferSpec,
227 replicationSpec: job.replicationSpec,
228 notificationConfig: job.notificationConfig,
229 loggingConfig: job.loggingConfig,
230 schedule: job.schedule,
231 eventStream: job.eventStream,
232 serviceAccount: job.serviceAccount,
233 creationTime: job.creationTime,
234 lastModificationTime: job.lastModificationTime,
235 latestOperationName: job.latestOperationName,
236 };
237};
238
239const replaceOnImmutable = (
240 news: TransferJobProps,
241 previous: TransferJobProps | undefined,
242 output:
243 | {
244 jobId?: string;
245 schedule?: Schedule;
246 eventStream?: EventStream;
247 serviceAccount?: string;
248 replicationSpec?: ReplicationSpec;
249 }
250 | undefined,
251) => {
252 const previousId = previous?.jobId ?? output?.jobId;
253 const nextId =
254 news.jobId !== undefined ? sanitizeJobId(news.jobId) : previousId;
255 if (
256 previousId !== undefined &&
257 nextId !== undefined &&
258 previousId !== nextId
259 ) {
260 return { action: "replace" as const, deleteFirst: false };
261 }
262 if (output === undefined && previous === undefined) return undefined;
263 const observedSchedule = previous?.schedule ?? output?.schedule;
264 const observedStream = previous?.eventStream ?? output?.eventStream;
265 const observedAccount = previous?.serviceAccount ?? output?.serviceAccount;
266 const observedReplication =
267 previous?.replicationSpec ?? output?.replicationSpec;
268 if (!sameValue(news.schedule, observedSchedule)) {
269 return { action: "replace" as const, deleteFirst: false };
270 }
271 if (!sameValue(news.eventStream, observedStream)) {
272 return { action: "replace" as const, deleteFirst: false };
273 }
274 if ((news.serviceAccount ?? "") !== (observedAccount ?? "")) {
275 return { action: "replace" as const, deleteFirst: false };
276 }
277 if (!sameValue(news.replicationSpec, observedReplication)) {
278 return { action: "replace" as const, deleteFirst: false };
279 }
280 return undefined;
281};
282
283const toCreateBody = (
284 name: string,
285 projectId: string,
286 news: TransferJobProps,
287 description: string,
288): storagetransfer.TransferJob => ({
289 name,
290 projectId,
291 description,
292 transferSpec: news.transferSpec,
293 replicationSpec: news.replicationSpec,
294 notificationConfig: news.notificationConfig,
295 loggingConfig: news.loggingConfig,
296 schedule: news.schedule,
297 eventStream: news.eventStream,
298 serviceAccount: news.serviceAccount,
299 status: statusOf(news.status),
300});
301
302export const TransferJobProvider = () =>
303 Provider.succeed(TransferJob, {
304 stables: ["name", "jobId", "project", "creationTime"],
305
306 diff: Effect.fn(function* ({ news, olds, output }) {
307 if (!isResolved(news)) return undefined;
308 return replaceOnImmutable(news, olds, output);
309 }),
310
311 read: Effect.fn(function* ({ id, olds, output }) {
312 const env = yield* GcpEnvironment.current;
313 const jobId = yield* toJobId(id, olds?.jobId, output?.jobId);
314 const name = output?.name ?? transferJobName(jobId);
315 const existing = yield* getTransferJob(name, env.project);
316 if (existing === undefined || isDeletedJob(existing)) return undefined;
317 const attrs = toAttrs(existing, env.project);
318 return (yield* ownedByAlchemy(id, existing.description))
319 ? attrs
320 : Unowned(attrs);
321 }),
322
323 list: () =>
324 Effect.gen(function* () {
325 const env = yield* GcpEnvironment.current;
326 const jobs = yield* listTransferJobs(env.project);
327 return jobs
328 .filter(
329 (job) => !isDeletedJob(job) && hasOwnershipMarker(job.description),
330 )
331 .map((job) => toAttrs(job, env.project));
332 }),
333
334 reconcile: Effect.fn(function* ({ id, news, output }) {
335 const env = yield* GcpEnvironment.current;
336 const jobId = yield* toJobId(id, news.jobId, output?.jobId);
337 const name = transferJobName(jobId);
338 const ownership = yield* createOwnership(id);
339 const desiredDescription = encodeOwnership(ownership, news.description);
340 const desiredStatus = statusOf(news.status);
341
342 let current = yield* getTransferJob(output?.name ?? name, env.project);
343
344 if (current === undefined) {
345 const created = yield* retryApiDisabled(
346 storagetransfer.createTransferJobs({
347 body: toCreateBody(name, env.project, news, desiredDescription),
348 }),
349 ).pipe(
350 Effect.catchTag("Conflict", () => getTransferJob(name, env.project)),
351 );
352 current = created ?? undefined;
353 }
354
355 if (current === undefined) {
356 return yield* new TransferJobNotResolved({ name });
357 }
358
359 if (isDeletedJob(current)) {
360 current = yield* storagetransfer.patchTransferJobs({
361 jobName: current.name ?? name,
362 body: {
363 projectId: env.project,
364 updateTransferJobFieldMask: fieldMask([
365 "description",
366 news.transferSpec !== undefined ? "transfer_spec" : undefined,
367 news.notificationConfig !== undefined
368 ? "notification_config"
369 : undefined,
370 news.loggingConfig !== undefined ? "logging_config" : undefined,
371 "status",
372 ]),
373 transferJob: {
374 description: desiredDescription,
375 transferSpec: news.transferSpec,
376 notificationConfig: news.notificationConfig,
377 loggingConfig: news.loggingConfig,
378 status: desiredStatus,
379 },
380 },
381 });
382 }
383
384 const descriptionChanged =
385 (current.description ?? "") !== desiredDescription;
386 const specChanged = !sameValue(current.transferSpec, news.transferSpec);
387 const notificationChanged = !sameValue(
388 current.notificationConfig,
389 news.notificationConfig,
390 );
391 const loggingChanged = !sameValue(
392 current.loggingConfig,
393 news.loggingConfig,
394 );
395 const statusChanged = (current.status ?? "") !== desiredStatus;
396 const updateMask = fieldMask([
397 descriptionChanged ? "description" : undefined,
398 specChanged && news.transferSpec !== undefined
399 ? "transfer_spec"
400 : undefined,
401 notificationChanged ? "notification_config" : undefined,
402 loggingChanged ? "logging_config" : undefined,
403 statusChanged ? "status" : undefined,
404 ]);
405
406 if (updateMask.length > 0) {
407 current = yield* storagetransfer.patchTransferJobs({
408 jobName: current.name ?? name,
409 body: {
410 projectId: env.project,
411 updateTransferJobFieldMask: updateMask,
412 transferJob: {
413 description: desiredDescription,
414 transferSpec: news.transferSpec,
415 notificationConfig: news.notificationConfig,
416 loggingConfig: news.loggingConfig,
417 status: desiredStatus,
418 },
419 },
420 });
421 }
422
423 return toAttrs(current, env.project);
424 }),
425
426 delete: Effect.fn(function* ({ output }) {
427 yield* storagetransfer
428 .deleteTransferJobs({
429 jobName: output.name,
430 projectId: output.project,
431 })
432 .pipe(Effect.catchTag("NotFound", () => Effect.void));
433 }),
434 });