1import * as translate from "@distilled.cloud/aws/translate";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13import { readTranslateTags, syncTranslateTags } from "./internal.ts";
14
15/**
16 * The parallel data import failed server-side (`Status: FAILED`) — usually a
17 * malformed input file or an S3 URI the caller cannot read.
18 */
19export class ParallelDataFailed extends Data.TaggedError("ParallelDataFailed")<{
20 readonly name: string;
21 readonly message: string | undefined;
22}> {}
23
24/**
25 * The parallel data did not reach a terminal status within the bounded
26 * polling window.
27 */
28export class ParallelDataNotConverged extends Data.TaggedError(
29 "ParallelDataNotConverged",
30)<{
31 readonly name: string;
32 readonly status: string | undefined;
33}> {}
34
35export interface ParallelDataProps {
36 /**
37 * Name of the parallel data — up to 256 characters of letters, digits, and
38 * hyphens (`^([A-Za-z0-9-]_?)+$`). If omitted, a unique name is generated
39 * from the app, stage, and logical ID. Changing the name replaces the
40 * parallel data.
41 */
42 parallelDataName?: string;
43 /**
44 * Description of the parallel data.
45 */
46 description?: string;
47 /**
48 * S3 URI of the parallel data input file (e.g.
49 * `s3://my-bucket/examples.csv`). The caller must be able to read the
50 * object. Updatable in place — a new import runs on change.
51 */
52 s3Uri: string;
53 /**
54 * Format of the input file: `TSV`, `CSV`, or `TMX`.
55 */
56 format?: "TSV" | "CSV" | "TMX";
57 /**
58 * Id of the customer-managed KMS key used to encrypt the parallel data.
59 * If omitted, Translate uses an AWS-owned key.
60 */
61 encryptionKeyId?: string;
62 /**
63 * User-defined tags for the parallel data.
64 */
65 tags?: Record<string, string>;
66}
67
68export interface ParallelData extends Resource<
69 "AWS.Translate.ParallelData",
70 ParallelDataProps,
71 {
72 /**
73 * Name of the parallel data — pass it as `ParallelDataNames` to
74 * `StartTextTranslationJob` for Active Custom Translation.
75 */
76 parallelDataName: string;
77 /**
78 * ARN of the parallel data, e.g.
79 * `arn:aws:translate:us-east-1:123456789012:parallel-data/style-examples`.
80 */
81 parallelDataArn: string;
82 /** Status of the parallel data (`ACTIVE` once import succeeds). */
83 status: string | undefined;
84 /** Source language code detected from the input file, e.g. `en`. */
85 sourceLanguageCode: string | undefined;
86 /** Target language codes detected from the input file. */
87 targetLanguageCodes: string[] | undefined;
88 /** Number of records imported. */
89 importedRecordCount: number | undefined;
90 /** Number of records that failed to import. */
91 failedRecordCount: number | undefined;
92 },
93 never,
94 Providers
95> {}
96
97/**
98 * Amazon Translate parallel data — segment-aligned translation examples
99 * imported from S3 that steer the style, tone, and word choice of batch
100 * translation jobs (Active Custom Translation). The import is asynchronous:
101 * the resource waits for the parallel data to become `ACTIVE`.
102 *
103 * ### Managing Parallel Data
104 * **Example:** Import parallel data from S3
105 * ```typescript
106 * const examples = yield* AWS.Translate.ParallelData("StyleExamples", {
107 * s3Uri: "s3://my-bucket/style-examples.csv",
108 * format: "CSV",
109 * });
110 * ```
111 *
112 * **Example:** Use parallel data in a batch translation job
113 * ```typescript
114 * const startJob = yield* AWS.Translate.StartTextTranslationJob(dataAccessRole);
115 * yield* startJob({
116 * InputDataConfig: { S3Uri: "s3://my-bucket/input/", ContentType: "text/plain" },
117 * OutputDataConfig: { S3Uri: "s3://my-bucket/output/" },
118 * SourceLanguageCode: "en",
119 * TargetLanguageCodes: ["es"],
120 * ParallelDataNames: [examples.parallelDataName],
121 * });
122 * ```
123 *
124 * @resource
125 */
126export const ParallelData = Resource<ParallelData>(
127 "AWS.Translate.ParallelData",
128);
129
130const toAttributes = (props: translate.ParallelDataProperties) => ({
131 parallelDataName: props.Name!,
132 parallelDataArn: props.Arn!,
133 status: props.Status,
134 sourceLanguageCode: props.SourceLanguageCode,
135 targetLanguageCodes: props.TargetLanguageCodes
136 ? [...props.TargetLanguageCodes]
137 : undefined,
138 importedRecordCount: props.ImportedRecordCount,
139 failedRecordCount: props.FailedRecordCount,
140});
141
142export const ParallelDataProvider = () =>
143 Provider.effect(
144 ParallelData,
145 Effect.gen(function* () {
146 const createName = Effect.fn(function* (
147 id: string,
148 props: ParallelDataProps,
149 ) {
150 return (
151 props.parallelDataName ??
152 (yield* createPhysicalName({ id, maxLength: 256 }))
153 );
154 });
155
156 const getOne = Effect.fn(function* (name: string) {
157 return yield* translate
158 .getParallelData({ Name: name })
159 .pipe(
160 Effect.catchTag("ResourceNotFoundException", () =>
161 Effect.succeed(undefined),
162 ),
163 );
164 });
165
166 // The import runs asynchronously (CREATING/UPDATING → ACTIVE|FAILED);
167 // poll bounded (~10 min ceiling — small files typically converge in a
168 // few minutes).
169 const waitForStable = Effect.fn(function* (name: string) {
170 const final = yield* getOne(name).pipe(
171 Effect.repeat({
172 schedule: Schedule.spaced("10 seconds"),
173 until: (r): boolean => {
174 const status = r?.ParallelDataProperties?.Status;
175 return (
176 r === undefined || status === "ACTIVE" || status === "FAILED"
177 );
178 },
179 times: 60,
180 }),
181 );
182 const props = final?.ParallelDataProperties;
183 if (props?.Status === "FAILED") {
184 return yield* Effect.fail(
185 new ParallelDataFailed({ name, message: props.Message }),
186 );
187 }
188 if (props?.Status !== "ACTIVE") {
189 return yield* Effect.fail(
190 new ParallelDataNotConverged({ name, status: props?.Status }),
191 );
192 }
193 return props;
194 });
195
196 return {
197 stables: ["parallelDataName", "parallelDataArn"],
198
199 diff: Effect.fn(function* ({ id, olds, news }) {
200 if (!isResolved(news)) return undefined;
201 if ((yield* createName(id, olds)) !== (yield* createName(id, news))) {
202 return { action: "replace" } as const;
203 }
204 }),
205
206 read: Effect.fn(function* ({ id, olds, output }) {
207 const name =
208 output?.parallelDataName ??
209 (yield* createName(id, olds ?? { s3Uri: "" }));
210 const found = yield* getOne(name);
211 if (found?.ParallelDataProperties === undefined) return undefined;
212 const attrs = toAttributes(found.ParallelDataProperties);
213 const tags = yield* readTranslateTags(attrs.parallelDataArn);
214 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
215 }),
216
217 reconcile: Effect.fn(function* ({ id, news, output, session }) {
218 const name =
219 output?.parallelDataName ?? (yield* createName(id, news));
220 const internalTags = yield* createInternalTags(id);
221 const desiredTags = { ...internalTags, ...news.tags };
222
223 // Observe — cloud state is authoritative.
224 const observed = yield* getOne(name);
225 const observedProps = observed?.ParallelDataProperties;
226
227 if (observedProps === undefined) {
228 // Ensure — greenfield import. ConflictException is a race with a
229 // concurrent create of the same name; fall through to observe.
230 // The client token only needs to be unique per submission;
231 // idempotency across crashed runs comes from observing first.
232 const clientToken = yield* Effect.sync(() => crypto.randomUUID());
233 yield* translate
234 .createParallelData({
235 Name: name,
236 ClientToken: clientToken,
237 Description: news.description,
238 ParallelDataConfig: { S3Uri: news.s3Uri, Format: news.format },
239 EncryptionKey: news.encryptionKeyId
240 ? { Type: "KMS", Id: news.encryptionKeyId }
241 : undefined,
242 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
243 Key,
244 Value,
245 })),
246 })
247 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
248 } else {
249 // Sync — diff the observed import config against the desired one
250 // and re-import only on drift.
251 const config = observedProps.ParallelDataConfig;
252 const drift =
253 config?.S3Uri !== news.s3Uri ||
254 (news.format !== undefined && config?.Format !== news.format) ||
255 (observedProps.Description ?? undefined) !==
256 (news.description ?? undefined);
257 if (drift) {
258 // Computed outside the retry so replays of the same submission
259 // reuse one idempotency token.
260 const clientToken = yield* Effect.sync(() => crypto.randomUUID());
261 yield* translate
262 .updateParallelData({
263 Name: name,
264 ClientToken: clientToken,
265 Description: news.description,
266 ParallelDataConfig: {
267 S3Uri: news.s3Uri,
268 Format: news.format,
269 },
270 })
271 .pipe(
272 // A concurrent import may still be settling — retry bounded.
273 Effect.retry({
274 while: (e): boolean =>
275 e._tag === "ConcurrentModificationException" ||
276 e._tag === "ConflictException",
277 schedule: Schedule.spaced("10 seconds"),
278 times: 8,
279 }),
280 );
281 }
282 }
283
284 yield* session.note(name);
285 const props = yield* waitForStable(name);
286 yield* syncTranslateTags(props.Arn!, desiredTags);
287 return toAttributes(props);
288 }),
289
290 delete: Effect.fn(function* ({ output }) {
291 // Idempotent — the parallel data may already be gone. A delete
292 // during CREATING/UPDATING conflicts; retry bounded.
293 yield* translate
294 .deleteParallelData({ Name: output.parallelDataName })
295 .pipe(
296 Effect.retry({
297 while: (e): boolean =>
298 e._tag === "ConcurrentModificationException",
299 schedule: Schedule.spaced("10 seconds"),
300 times: 8,
301 }),
302 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
303 );
304 // Deletion is asynchronous (DELETING) — wait until gone so a
305 // replacement can immediately reuse the name.
306 yield* translate
307 .getParallelData({ Name: output.parallelDataName })
308 .pipe(
309 Effect.catchTag("ResourceNotFoundException", () =>
310 Effect.succeed(undefined),
311 ),
312 Effect.repeat({
313 schedule: Schedule.spaced("5 seconds"),
314 until: (r): boolean => r === undefined,
315 times: 36,
316 }),
317 );
318 }),
319
320 list: () =>
321 translate.listParallelData.pages({}).pipe(
322 Stream.runCollect,
323 Effect.map((chunk) =>
324 Array.from(chunk)
325 .flatMap((page) => page.ParallelDataPropertiesList ?? [])
326 .filter((props) => props.Name && props.Arn)
327 .map(toAttributes),
328 ),
329 ),
330 };
331 }),
332 );