clover's git

the-knife

1	import type * as DynamoDB from "@distilled.cloud/aws/dynamodb";
2	import type { TimeToLiveSpecification } from "@distilled.cloud/aws/dynamodb";
3	import * as cloudwatch from "@distilled.cloud/aws/cloudwatch";
4	import * as dynamodb from "@distilled.cloud/aws/dynamodb";
5	import type * as lambda from "aws-lambda";
6	import * as Data from "effect/Data";
7	import type * as Duration from "effect/Duration";
8	import * as Effect from "effect/Effect";
9	import * as Schedule from "effect/Schedule";
10	import * as Stream from "effect/Stream";
11
12	import { Unowned } from "../../AdoptPolicy.ts";
13	import { havePropsChanged, isResolved } from "../../Diff.ts";
14	import type { Input } from "../../Input.ts";
15	import { createPhysicalName } from "../../PhysicalName.ts";
16	import * as Provider from "../../Provider.ts";
17	import { Resource } from "../../Resource.ts";
18	import {
19	createInternalTags,
20	createTagsList,
21	diffTags,
22	hasAlchemyTags,
23	} from "../../Tags.ts";
24	import { toWireDays } from "../../Util/Duration.ts";
25	import type { AccountID } from "../Environment.ts";
26	import type { Providers } from "../Providers.ts";
27	import type { RegionID } from "../Region.ts";
28
29	export type TableName = string;
30
31	export type TableArn =
32	`arn:aws:dynamodb:${RegionID}:${AccountID}:table/${TableName}`;
33
34	export type TableRecord<Data> = Omit<lambda.DynamoDBRecord, "dynamodb"> & {
35	dynamodb: Omit<lambda.StreamRecord, "NewImage" | "OldImage"> & {
36	NewImage?: Data;
37	OldImage?: Data;
38	};
39	};
40
41	export type TableEvent<Data> = Omit<lambda.DynamoDBStreamEvent, "Records"> & {
42	Records: TableRecord<Data>[];
43	};
44
45	export type ScalarAttributeType = "S" | "N" | "B";
46
47	export interface PointInTimeRecoverySpecification {
48	/**
49	* Whether point-in-time recovery (continuous backups) is enabled for the
50	* table.
51	*/
52	pointInTimeRecoveryEnabled: boolean;
53	/**
54	* How far back the table can be restored, e.g. `"7 days"` or
55	* `Duration.days(7)`. Rounded to whole days on the wire (1–35 days).
56	* @default 35 days
57	*/
58	recoveryPeriod?: Duration.Input;
59	}
60
61	export interface KinesisStreamingDestination {
62	/**
63	* ARN of the destination Kinesis Data Stream. The stream must be ACTIVE
64	* when the destination is enabled (deploy the `AWS.Kinesis.Stream`
65	* resource in the same stack and pass `stream.streamArn`).
66	*/
67	streamArn: string;
68	/**
69	* Precision of the approximate creation timestamp stamped on records
70	* delivered to the Kinesis stream. Changing this on an ACTIVE destination
71	* updates it in place, but the UPDATING→ACTIVE transition can take several
72	* minutes.
73	* @default "MILLISECOND"
74	*/
75	approximateCreationDateTimePrecision?: DynamoDB.ApproximateCreationDateTimePrecision;
76	}
77
78	/**
79	* One or more attribute names forming an index key. DynamoDB's wire format
80	* flattens both key segments into a single ordered `KeySchema` list, but
81	* semantically a key is two ordered segments — this type captures one of
82	* them. Order is significant: partition attributes are hashed together in
83	* declaration order, and sort attributes are queried left-to-right.
84	*/
85	export type IndexKey = string | string[];
86
87	export interface LocalSecondaryIndexProps {
88	/**
89	* Name of the index, unique within the table.
90	*/
91	indexName: string;
92	/**
93	* Sort key attribute name. An LSI always shares the table's partition
94	* key, so only the sort key is declared; multi-attribute keys are not
95	* supported on LSIs.
96	*/
97	sortKey: string;
98	/**
99	* Attributes projected from the table into the index.
100	*/
101	projection: DynamoDB.Projection;
102	}
103
104	export interface GlobalSecondaryIndexProps {
105	/**
106	* Name of the index, unique within the table.
107	*/
108	indexName: string;
109	/**
110	* Partition key attribute name(s). Up to four attributes may be listed;
111	* they are hashed together in declaration order and every one must be
112	* specified with an equality condition when querying the index.
113	*/
114	partitionKey: IndexKey;
115	/**
116	* Optional sort key attribute name(s). Up to four attributes may be
117	* listed; items sort by each attribute in declaration order and queries
118	* narrow them left-to-right (no gaps, inequality last).
119	*/
120	sortKey?: IndexKey;
121	/**
122	* Attributes projected from the table into the index.
123	*/
124	projection: DynamoDB.Projection;
125	provisionedThroughput?: DynamoDB.ProvisionedThroughput;
126	onDemandThroughput?: DynamoDB.OnDemandThroughput;
127	warmThroughput?: DynamoDB.WarmThroughput;
128	}
129
130	export type TableProps = {
131	/**
132	* Name of the table. If omitted, Alchemy generates a deterministic physical
133	* name from the stack, stage, and logical ID, suffixed with the instance ID
134	* so replacements can create the successor table before deleting the old one.
135	*
136	* Changing this property replaces the table. If the name is hard-coded,
137	* replacement-triggering changes (key schema, attribute types, LSIs, GSI
138	* key changes) delete the old table first and then recreate it under the
139	* same name, since DynamoDB table names are unique per account/region.
140	*/
141	tableName?: string;
142	/**
143	* Partition key attribute name for the table.
144	*
145	* Base-table primary keys are always a single partition attribute plus an
146	* optional single sort attribute — DynamoDB supports multi-attribute keys
147	* only on global secondary indexes (see `globalSecondaryIndexes`).
148	*/
149	partitionKey: string;
150	/**
151	* Optional sort key attribute name for the table.
152	*/
153	sortKey?: string;
154	/**
155	* Attribute definitions used by the primary key and any secondary indexes.
156	*/
157	attributes: Record<string, ScalarAttributeType>;
158	/**
159	* Local secondary indexes, created with the table. An LSI always shares
160	* the table's partition key and declares a single sort key. Changing this
161	* property replaces the table.
162	*/
163	localSecondaryIndexes?: LocalSecondaryIndexProps[];
164	/**
165	* Global secondary indexes. GSIs support multi-attribute keys: up to four
166	* partition attributes (hashed together as the composite partition key)
167	* and up to four sort attributes (sorted and queried left-to-right).
168	* Attribute order is significant — reordering defines a different index
169	* and replaces the table. Every key attribute must appear in `attributes`.
170	*/
171	globalSecondaryIndexes?: GlobalSecondaryIndexProps[];
172	billingMode?: DynamoDB.BillingMode;
173	deletionProtectionEnabled?: boolean;
174	onDemandThroughput?: DynamoDB.OnDemandThroughput;
175	/**
176	* Enables point-in-time recovery (continuous backups) and optionally sets
177	* the recovery window via `recoveryPeriod` (a `Duration.Input`, rounded to
178	* whole days on the wire).
179	*/
180	pointInTimeRecoverySpecification?: PointInTimeRecoverySpecification;
181	provisionedThroughput?: DynamoDB.ProvisionedThroughput;
182	sseSpecification?: DynamoDB.SSESpecification;
183	tags?: Record<string, string>;
184	timeToLiveSpecification?: DynamoDB.TimeToLiveSpecification;
185	warmThroughput?: DynamoDB.WarmThroughput;
186	tableClass?: DynamoDB.TableClass;
187	/**
188	* Resource-based IAM policy document (JSON string) attached to the table.
189	* Grants cross-account or scoped-principal access to the table and its
190	* indexes. Removing the property deletes the policy.
191	*/
192	resourcePolicy?: string;
193	/**
194	* Streams change data capture records to the given Kinesis Data Stream.
195	* Changing the stream disables the old destination before enabling the
196	* new one; removing the property disables streaming.
197	*/
198	kinesisStreamingDestination?: KinesisStreamingDestination;
199	/**
200	* Enables CloudWatch Contributor Insights (table-level rule) for the
201	* table, surfacing the most-accessed and most-throttled keys.
202	* @default false
203	*/
204	contributorInsightsEnabled?: boolean;
205	};
206
207	export type TableBinding = {
208	streamSpecification?: DynamoDB.StreamSpecification;
209	};
210
211	export interface Table extends Resource<
212	"AWS.DynamoDB.Table",
213	TableProps,
214	{
215	/** The unique ID AWS assigns to the table. */
216	tableId: string;
217	/** The physical name of the table. */
218	tableName: TableName;
219	/** The ARN of the table. */
220	tableArn: TableArn;
221	/** The partition (hash) key attribute name. */
222	partitionKey: string;
223	/** The sort (range) key attribute name, if defined. */
224	sortKey: string | undefined;
225	/** ARN of the most recent DynamoDB stream (when streams are enabled). */
226	latestStreamArn: string | undefined;
227	/** The current stream configuration (when streams are enabled). */
228	streamSpecification: DynamoDB.StreamSpecification | undefined;
229	/** Descriptions of the table's local secondary indexes. */
230	localSecondaryIndexes:
231	| DynamoDB.LocalSecondaryIndexDescription[]
232	| undefined;
233	/** Descriptions of the table's global secondary indexes. */
234	globalSecondaryIndexes:
235	| DynamoDB.GlobalSecondaryIndexDescription[]
236	| undefined;
237	/** The point-in-time recovery status of the table. */
238	pointInTimeRecoveryDescription:
239	| DynamoDB.PointInTimeRecoveryDescription
240	| undefined;
241	/** The tags attached to the table. */
242	tags: Record<string, string> | undefined;
243	},
244	TableBinding,
245	Providers
246	> {}
247
248	/**
249	* An Amazon DynamoDB table with optional indexes, PITR, TTL, and stream-aware
250	* binding support.
251	*
252	* `Table` owns the lifecycle of the physical table while the binding contract
253	* allows runtime-specific integrations such as Lambda table event sources to
254	* request stream configuration without forcing a circular input prop.
255	* ### Creating Tables
256	* **Example:** Basic Table
257	* ```typescript
258	* import * as DynamoDB from "alchemy/AWS/DynamoDB";
259	*
260	* const table = yield* DynamoDB.Table("UsersTable", {
261	* partitionKey: "pk",
262	* attributes: {
263	* pk: "S",
264	* },
265	* });
266	* ```
267	*
268	* **Example:** Table with Sort Key and TTL
269	* ```typescript
270	* const table = yield* DynamoDB.Table("SessionsTable", {
271	* partitionKey: "userId",
272	* sortKey: "sessionId",
273	* attributes: {
274	* userId: "S",
275	* sessionId: "S",
276	* expiresAt: "N",
277	* },
278	* timeToLiveSpecification: {
279	* Enabled: true,
280	* AttributeName: "expiresAt",
281	* },
282	* });
283	* ```
284	*
285	* **Example:** Table with Global Secondary Index
286	* ```typescript
287	* const table = yield* DynamoDB.Table("OrdersTable", {
288	* partitionKey: "pk",
289	* sortKey: "sk",
290	* attributes: {
291	* pk: "S",
292	* sk: "S",
293	* gsi1pk: "S",
294	* gsi1sk: "S",
295	* },
296	* globalSecondaryIndexes: [{
297	* indexName: "GSI1",
298	* partitionKey: "gsi1pk",
299	* sortKey: "gsi1sk",
300	* projection: { ProjectionType: "ALL" },
301	* }],
302	* });
303	* ```
304	*
305	* **Example:** Multi-Attribute GSI Keys
306	* GSI partition and sort keys may be composed of up to four attributes each,
307	* indexing natural domain attributes directly instead of synthetic
308	* concatenated keys. Partition attributes are hashed together (queries must
309	* specify all of them with equality); sort attributes are queried
310	* left-to-right in declaration order.
311	* ```typescript
312	* const matches = yield* DynamoDB.Table("TournamentMatches", {
313	* partitionKey: "matchId",
314	* attributes: {
315	* matchId: "S",
316	* tournamentId: "S",
317	* region: "S",
318	* round: "S",
319	* },
320	* globalSecondaryIndexes: [{
321	* indexName: "TournamentRegionIndex",
322	* partitionKey: ["tournamentId", "region"],
323	* sortKey: ["round", "matchId"],
324	* projection: { ProjectionType: "ALL" },
325	* }],
326	* });
327	*
328	* // init
329	* const query = yield* AWS.DynamoDB.Query(matches);
330	*
331	* // runtime: query with every partition attribute, then narrow the sort
332	* // attributes left-to-right
333	* const response = yield* query({
334	* IndexName: "TournamentRegionIndex",
335	* KeyConditionExpression:
336	* "tournamentId = :t AND #r = :r AND round = :round",
337	* ExpressionAttributeNames: { "#r": "region" },
338	* ExpressionAttributeValues: {
339	* ":t": { S: "WINTER2024" },
340	* ":r": { S: "NA-EAST" },
341	* ":round": { S: "SEMIFINALS" },
342	* },
343	* });
344	* ```
345	*
346	* ### Runtime Operations
347	* Bind DynamoDB operations in the init phase and use them in runtime
348	* handlers. Bindings inject the table name and grant scoped IAM
349	* permissions automatically.
350	*
351	* **Example:** Read and write items
352	* ```typescript
353	* // init
354	* const getItem = yield* AWS.DynamoDB.GetItem(table);
355	* const putItem = yield* AWS.DynamoDB.PutItem(table);
356	*
357	* return {
358	* fetch: Effect.gen(function* () {
359	* // runtime
360	* yield* putItem({
361	* Item: { pk: { S: "user#123" }, name: { S: "Alice" } },
362	* });
363	* const result = yield* getItem({
364	* Key: { pk: { S: "user#123" } },
365	* });
366	* return yield* HttpServerResponse.json(result.Item);
367	* }),
368	* };
369	* ```
370	*
371	* ### Table Features
372	* **Example:** Resource Policy
373	* ```typescript
374	* const table = yield* DynamoDB.Table("SharedTable", {
375	* partitionKey: "pk",
376	* attributes: { pk: "S" },
377	* resourcePolicy: JSON.stringify({
378	* Version: "2012-10-17",
379	* Statement: [{
380	* Effect: "Allow",
381	* Principal: { AWS: "arn:aws:iam::111122223333:root" },
382	* Action: ["dynamodb:GetItem", "dynamodb:Query"],
383	* Resource: "*",
384	* }],
385	* }),
386	* });
387	* ```
388	*
389	* **Example:** Kinesis Streaming Destination
390	* ```typescript
391	* import * as Kinesis from "alchemy/AWS/Kinesis";
392	*
393	* const stream = yield* Kinesis.Stream("CdcStream", {});
394	* const table = yield* DynamoDB.Table("CdcTable", {
395	* partitionKey: "pk",
396	* attributes: { pk: "S" },
397	* kinesisStreamingDestination: {
398	* streamArn: stream.streamArn,
399	* approximateCreationDateTimePrecision: "MICROSECOND",
400	* },
401	* });
402	* ```
403	*
404	* **Example:** Contributor Insights
405	* ```typescript
406	* const table = yield* DynamoDB.Table("HotKeyTable", {
407	* partitionKey: "pk",
408	* attributes: { pk: "S" },
409	* contributorInsightsEnabled: true,
410	* });
411	* ```
412	*
413	* ### DynamoDB Streams
414	* Process change data capture events from a DynamoDB table using a
415	* Lambda event source mapping. The stream is enabled automatically
416	* through the binding contract.
417	*
418	* **Example:** Process table changes
419	* ```typescript
420	* // init
421	* yield* DynamoDB.consumeTableChanges(
422	* table,
423	* { streamViewType: "NEW_AND_OLD_IMAGES" },
424	* Effect.fn(function* (record) {
425	* yield* Effect.log(`${record.eventName}: ${JSON.stringify(record.dynamodb)}`);
426	* }),
427	* );
428	* ```
429	*
430	* @resource
431	*/
432	export const Table = Resource<Table>("AWS.DynamoDB.Table");
433
434	export const TableProvider = () =>
435	Provider.effect(
436	Table,
437	Effect.gen(function* () {
438	const createTableName = (
439	id: string,
440	props: Input.ResolveProps<TableProps>,
441	) =>
442	Effect.gen(function* () {
443	return (
444	props.tableName ??
445	(yield* createPhysicalName({
446	id,
447	// see: https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_TableDescription.html#DDB-Type-TableDescription-TableName
448	maxLength: 255,
449	}))
450	);
451	});
452
453	const toKeySchema = (props: Input.ResolveProps<TableProps>) => [
454	{
455	AttributeName: props.partitionKey,
456	KeyType: "HASH" as const,
457	},
458	...(props.sortKey
459	? [
460	{
461	AttributeName: props.sortKey,
462	KeyType: "RANGE" as const,
463	},
464	]
465	: []),
466	];
467
468	const toKeyAttributeNames = (key: IndexKey | undefined) =>
469	key === undefined ? [] : typeof key === "string" ? [key] : key;
470
471	// AWS's wire format flattens an index key into one ordered KeySchema
472	// list and validates that all HASH elements precede all RANGE
473	// elements; deriving the list from the two typed segments makes the
474	// invalid orderings unrepresentable in props.
475	const toIndexKeySchema = (
476	partitionKey: IndexKey,
477	sortKey: IndexKey | undefined,
478	): DynamoDB.KeySchemaElement[] => [
479	...toKeyAttributeNames(partitionKey).map((name) => ({
480	AttributeName: name,
481	KeyType: "HASH" as const,
482	})),
483	...toKeyAttributeNames(sortKey).map((name) => ({
484	AttributeName: name,
485	KeyType: "RANGE" as const,
486	})),
487	];
488
489	// Pre-typed-props state may persist index props in the legacy wire
490	// shape ({ IndexName, KeySchema, ... }). Tolerate it when converting
491	// `olds` so upgrading alchemy never plans a spurious table
492	// replacement over a shape-only difference.
493	const isLegacyWireIndex = (
494	index: unknown,
495	): index is DynamoDB.GlobalSecondaryIndex =>
496	typeof index === "object" && index !== null && "KeySchema" in index;
497
498	const toWireGlobalSecondaryIndex = (
499	index: GlobalSecondaryIndexProps,
500	): DynamoDB.GlobalSecondaryIndex =>
501	isLegacyWireIndex(index)
502	? index
503	: {
504	IndexName: index.indexName,
505	KeySchema: toIndexKeySchema(index.partitionKey, index.sortKey),
506	Projection: index.projection,
507	ProvisionedThroughput: index.provisionedThroughput,
508	OnDemandThroughput: index.onDemandThroughput,
509	WarmThroughput: index.warmThroughput,
510	};
511
512	const toWireLocalSecondaryIndex = (
513	tablePartitionKey: string,
514	index: LocalSecondaryIndexProps,
515	): DynamoDB.LocalSecondaryIndex =>
516	isLegacyWireIndex(index)
517	? index
518	: {
519	IndexName: index.indexName,
520	KeySchema: toIndexKeySchema(tablePartitionKey, index.sortKey),
521	Projection: index.projection,
522	};
523
524	const toWireGlobalSecondaryIndexes = (
525	indexes: readonly GlobalSecondaryIndexProps[] | undefined,
526	) =>
527	indexes === undefined || indexes.length === 0
528	? undefined
529	: indexes.map(toWireGlobalSecondaryIndex);
530
531	const toWireLocalSecondaryIndexes = (
532	tablePartitionKey: string,
533	indexes: readonly LocalSecondaryIndexProps[] | undefined,
534	) =>
535	indexes === undefined || indexes.length === 0
536	? undefined
537	: indexes.map((index) =>
538	toWireLocalSecondaryIndex(tablePartitionKey, index),
539	);
540
541	const toAttributeDefinitions = (
542	attrs: Record<string, ScalarAttributeType>,
543	) =>
544	Object.entries(attrs)
545	.map(([name, type]) => ({
546	AttributeName: name,
547	AttributeType: type,
548	}))
549	.sort((a, b) => a.AttributeName.localeCompare(b.AttributeName));
550
551	const adoptExistingTable = (tableName: string) =>
552	// The engine has cleared us via `read` (foreign-tagged tables are
553	// surfaced as `Unowned`). On a race between read and create, just
554	// describe the existing table and continue.
555	dynamodb
556	.describeTable({ TableName: tableName })
557	.pipe(Effect.map((r) => r.Table!));
558
559	const createTags = Effect.fn(function* (
560	id: string,
561	tags?: Record<string, string>,
562	) {
563	return {
564	...(yield* createInternalTags(id)),
565	...tags,
566	};
567	});
568
569	const normalizeStreamSpecification = (
570	streamSpecification: DynamoDB.StreamSpecification | undefined,
571	) =>
572	streamSpecification?.StreamEnabled === true
573	? ({
574	StreamEnabled: true,
575	StreamViewType: streamSpecification.StreamViewType,
576	} satisfies DynamoDB.StreamSpecification)
577	: undefined;
578
579	const resolveStreamSpecification = (
580	bindings: ReadonlyArray<TableBinding | { data?: TableBinding }>,
581	) =>
582	Effect.gen(function* () {
583	const requested = bindings
584	.flatMap((binding) =>
585	(binding as { data?: TableBinding }).data?.streamSpecification
586	?.StreamEnabled === true
587	? [
588	normalizeStreamSpecification(
589	(binding as { data?: TableBinding }).data
590	?.streamSpecification,
591	),
592	]
593	: (binding as TableBinding).streamSpecification
594	?.StreamEnabled === true
595	? [
596	normalizeStreamSpecification(
597	(binding as TableBinding).streamSpecification,
598	),
599	]
600	: [],
601	)
602	.filter((spec) => spec !== undefined);
603
604	if (requested.length === 0) {
605	return undefined;
606	}
607
608	const [first, ...rest] = requested;
609	if (!first?.StreamViewType) {
610	return yield* Effect.fail(new MissingStreamViewType());
611	}
612
613	for (const spec of rest) {
614	if (spec.StreamViewType !== first.StreamViewType) {
615	return yield* Effect.fail(
616	new ConflictingStreamViewTypes({
617	requested: requested.map((item) => item.StreamViewType),
618	}),
619	);
620	}
621	}
622
623	return first;
624	});
625
626	const isRetryableControlPlaneError = (error: { _tag?: string }) =>
627	error._tag === "InternalServerError" ||
628	error._tag === "LimitExceededException" ||
629	error._tag === "ResourceInUseException" ||
630	error._tag === "ResourceNotFoundException" ||
631	// Throttling: DynamoDB's control-plane API limits are low and the
632	// blanket SDK retry (10 attempts) is easily exhausted when the whole
633	// suite hammers create/update/describe concurrently. Give it the
634	// provider's much larger budget too.
635	error._tag === "ThrottlingException" ||
636	error._tag === "ProvisionedThroughputExceededException" ||
637	error._tag === "RequestLimitExceeded";
638
639	const waitForControlPlaneConvergence = Schedule.max([
640	Schedule.fixed("1 second"),
641	Schedule.recurs(120),
642	]);
643
644	const waitForTableActivationConvergence = Schedule.max([
645	Schedule.fixed("10 seconds"),
646	Schedule.recurs(180),
647	]);
648
649	const waitForGlobalSecondaryIndexesConvergence = Schedule.max([
650	Schedule.fixed("10 seconds"),
651	Schedule.recurs(180),
652	]);
653
654	const waitForDeletionConvergence = Schedule.max([
655	Schedule.fixed("1 second"),
656	Schedule.recurs(90),
657	]);
658
659	const formatPollingElapsed = (elapsedSeconds: number) =>
660	`${elapsedSeconds}s elapsed`;
661
662	const formatGlobalSecondaryIndexStatuses = (
663	indexes:
664	| readonly DynamoDB.GlobalSecondaryIndexDescription[]
665	| undefined,
666	) =>
667	JSON.stringify(
668	(indexes ?? []).map((index) => ({
669	name: index.IndexName,
670	status: index.IndexStatus,
671	backfilling: index.Backfilling,
672	})),
673	);
674
675	const updateTimeToLive = (
676	tableName: string,
677	timeToLiveSpecification: TimeToLiveSpecification,
678	) =>
679	dynamodb
680	.updateTimeToLive({
681	TableName: tableName,
682	TimeToLiveSpecification: timeToLiveSpecification!,
683	})
684	.pipe(
685	Effect.retry({
686	while: isRetryableControlPlaneError,
687	schedule: Schedule.max([
688	Schedule.exponential(100),
689	Schedule.recurs(30),
690	]),
691	}),
692	);
693
694	const updateContinuousBackups = (
695	tableName: string,
696	pointInTimeRecoverySpecification: DynamoDB.PointInTimeRecoverySpecification,
697	) =>
698	dynamodb
699	.updateContinuousBackups({
700	TableName: tableName,
701	PointInTimeRecoverySpecification: pointInTimeRecoverySpecification,
702	})
703	.pipe(
704	Effect.retry({
705	while: (e) =>
706	e._tag === "ContinuousBackupsUnavailableException" ||
707	isRetryableControlPlaneError(e),
708	schedule: Schedule.max([
709	Schedule.exponential(250),
710	Schedule.recurs(30),
711	]),
712	}),
713	);
714
715	// Resource policies are compared as canonicalized JSON so formatting
716	// differences between what the user wrote and what AWS stores don't
717	// cause phantom putResourcePolicy calls on every reconcile.
718	const canonicalPolicyDocument = (policy: string | undefined) => {
719	if (policy === undefined) return undefined;
720	try {
721	return JSON.stringify(JSON.parse(policy));
722	} catch {
723	return policy;
724	}
725	};
726
727	const readTableResourcePolicy = (tableArn: string) =>
728	dynamodb.getResourcePolicy({ ResourceArn: tableArn }).pipe(
729	Effect.map((response) => response.Policy),
730	Effect.catchTag("PolicyNotFoundException", () =>
731	Effect.succeed(undefined),
732	),
733	Effect.retry({
734	while: isRetryableControlPlaneError,
735	schedule: Schedule.max([
736	Schedule.exponential(250),
737	Schedule.recurs(15),
738	]),
739	}),
740	);
741
742	const putTableResourcePolicy = (tableArn: string, policy: string) =>
743	dynamodb
744	.putResourcePolicy({ ResourceArn: tableArn, Policy: policy })
745	.pipe(
746	Effect.retry({
747	while: isRetryableControlPlaneError,
748	schedule: Schedule.max([
749	Schedule.exponential(250),
750	Schedule.recurs(15),
751	]),
752	}),
753	);
754
755	const deleteTableResourcePolicy = (tableArn: string) =>
756	dynamodb.deleteResourcePolicy({ ResourceArn: tableArn }).pipe(
757	Effect.catchTag("PolicyNotFoundException", () => Effect.void),
758	Effect.retry({
759	while: isRetryableControlPlaneError,
760	schedule: Schedule.max([
761	Schedule.exponential(250),
762	Schedule.recurs(15),
763	]),
764	}),
765	);
766
767	const isKinesisDestinationTransitioning = (
768	destination: DynamoDB.KinesisDataStreamDestination,
769	) =>
770	destination.DestinationStatus === "ENABLING" ||
771	destination.DestinationStatus === "DISABLING" ||
772	destination.DestinationStatus === "UPDATING";
773
774	// Kinesis streaming destinations transition ENABLING→ACTIVE and
775	// DISABLING→DISABLED within seconds, but a precision change
776	// (UPDATING→ACTIVE) takes multiple minutes; poll bounded to 8 minutes.
777	const waitForKinesisDestinationsConvergence = Schedule.max([
778	Schedule.fixed("5 seconds"),
779	Schedule.recurs(96),
780	]);
781
782	const waitForKinesisDestinationsSettled = (
783	session: {
784	note: (message: string) => Effect.Effect<void, never, never>;
785	},
786	tableName: string,
787	) => {
788	let elapsedSeconds = 0;
789	return Effect.gen(function* () {
790	const response = yield* dynamodb.describeKinesisStreamingDestination({
791	TableName: tableName,
792	});
793	const destinations = response.KinesisDataStreamDestinations ?? [];
794	if (destinations.some(isKinesisDestinationTransitioning)) {
795	return yield* Effect.fail(new KinesisDestinationNotSettled());
796	}
797	return destinations;
798	}).pipe(
799	Effect.retry({
800	while: (error) =>
801	error._tag === "KinesisDestinationNotSettled" ||
802	isRetryableControlPlaneError(error),
803	schedule: waitForKinesisDestinationsConvergence.pipe(
804	Schedule.tap(({ attempt }) => {
805	elapsedSeconds = (attempt + 1) * 5;
806	return session.note(
807	`DynamoDB Table provider: waiting for Kinesis streaming destinations on ${tableName} to settle (${formatPollingElapsed(elapsedSeconds)})`,
808	);
809	}),
810	),
811	}),
812	);
813	};
814
815	const disableKinesisDestination = (
816	tableName: string,
817	streamArn: string,
818	) =>
819	dynamodb
820	.disableKinesisStreamingDestination({
821	TableName: tableName,
822	StreamArn: streamArn,
823	})
824	.pipe(
825	Effect.retry({
826	while: isRetryableControlPlaneError,
827	schedule: Schedule.max([
828	Schedule.exponential(250),
829	Schedule.recurs(15),
830	]),
831	}),
832	);
833
834	const enableKinesisDestination = (
835	tableName: string,
836	destination: KinesisStreamingDestination,
837	) =>
838	dynamodb
839	.enableKinesisStreamingDestination({
840	TableName: tableName,
841	StreamArn: destination.streamArn,
842	EnableKinesisStreamingConfiguration:
843	destination.approximateCreationDateTimePrecision !== undefined
844	? {
845	ApproximateCreationDateTimePrecision:
846	destination.approximateCreationDateTimePrecision,
847	}
848	: undefined,
849	})
850	.pipe(
851	Effect.retry({
852	while: isRetryableControlPlaneError,
853	schedule: Schedule.max([
854	Schedule.exponential(250),
855	Schedule.recurs(15),
856	]),
857	}),
858	);
859
860	const updateKinesisDestinationPrecision = (
861	tableName: string,
862	destination: KinesisStreamingDestination,
863	) =>
864	dynamodb
865	.updateKinesisStreamingDestination({
866	TableName: tableName,
867	StreamArn: destination.streamArn,
868	UpdateKinesisStreamingConfiguration: {
869	ApproximateCreationDateTimePrecision:
870	destination.approximateCreationDateTimePrecision,
871	},
872	})
873	.pipe(
874	Effect.retry({
875	while: isRetryableControlPlaneError,
876	schedule: Schedule.max([
877	Schedule.exponential(250),
878	Schedule.recurs(15),
879	]),
880	}),
881	);
882
883	// Contributor Insights transitions ENABLING→ENABLED / DISABLING→DISABLED
884	// within seconds, but UpdateContributorInsights rejects a DISABLE while
885	// the status is still ENABLING — wait for a settled status before
886	// diffing observed against desired.
887	const waitForContributorInsightsSettled = (
888	session: {
889	note: (message: string) => Effect.Effect<void, never, never>;
890	},
891	tableName: string,
892	) =>
893	Effect.gen(function* () {
894	const response = yield* dynamodb.describeContributorInsights({
895	TableName: tableName,
896	});
897	const status = response.ContributorInsightsStatus ?? "DISABLED";
898	if (status === "ENABLING" || status === "DISABLING") {
899	return yield* Effect.fail(new ContributorInsightsNotSettled());
900	}
901	return status;
902	}).pipe(
903	Effect.retry({
904	while: (error) =>
905	error._tag === "ContributorInsightsNotSettled" ||
906	isRetryableControlPlaneError(error),
907	schedule: Schedule.max([
908	Schedule.fixed("2 seconds"),
909	Schedule.recurs(45),
910	]).pipe(
911	Schedule.tap(({ attempt }) =>
912	session.note(
913	`DynamoDB Table provider: waiting for Contributor Insights on ${tableName} to settle (${formatPollingElapsed((attempt + 1) * 2)})`,
914	),
915	),
916	),
917	}),
918	);
919
920	// Enabling Contributor Insights makes DynamoDB create CloudWatch rules
921	// named `DynamoDBContributorInsights-{PKC|PKT}-<tableName>-<timestamp>`
922	// in the customer's account. Those rules can ONLY be removed by
923	// DynamoDB's own asynchronous cleanup, which runs when an
924	// UpdateContributorInsights DISABLE completes — CloudWatch's
925	// DeleteInsightRules/PutInsightRule reject them with AccessDenied even
926	// for account admins, and deleting the table aborts the cleanup and
927	// strands the rules forever (recoverable only via AWS support). The
928	// delete lifecycle therefore observes the rules and holds off
929	// deleteTable until DynamoDB's cleanup has actually removed them.
930	const listContributorInsightsRules = (tableName: string) =>
931	cloudwatch.describeInsightRules.pages({}).pipe(
932	Stream.runFold(
933	() => [] as string[],
934	(acc, page) => [
935	...acc,
936	...(page.InsightRules ?? [])
937	.map((rule) => rule.Name)
938	.filter(
939	(name): name is string =>
940	name !== undefined &&
941	name.startsWith("DynamoDBContributorInsights-") &&
942	name.includes(`-${tableName}-`),
943	),
944	],
945	),
946	);
947
948	const waitForContributorInsightsRulesDeleted = (
949	session: {
950	note: (message: string) => Effect.Effect<void, never, never>;
951	},
952	tableName: string,
953	) =>
954	Effect.gen(function* () {
955	const remaining = yield* listContributorInsightsRules(tableName);
956	if (remaining.length > 0) {
957	return yield* Effect.fail(new ContributorInsightsNotSettled());
958	}
959	}).pipe(
960	Effect.retry({
961	while: (error) =>
962	error._tag === "ContributorInsightsNotSettled" ||
963	isRetryableControlPlaneError(error),
964	schedule: Schedule.max([
965	Schedule.fixed("2 seconds"),
966	Schedule.recurs(20),
967	]).pipe(
968	Schedule.tap(({ attempt }) =>
969	session.note(
970	`DynamoDB Table provider: waiting for Contributor Insights rules of ${tableName} to be cleaned up (${formatPollingElapsed((attempt + 1) * 2)})`,
971	),
972	),
973	),
974	}),
975	// Rules from a PREVIOUS enable session whose cleanup was already
976	// aborted (e.g. by an out-of-band table deletion) can never be
977	// removed by anyone but AWS support — don't wedge the delete on
978	// them; surface a note and move on.
979	Effect.catchTag("ContributorInsightsNotSettled", () =>
980	session.note(
981	`DynamoDB Table provider: Contributor Insights rules for ${tableName} were not cleaned up by DynamoDB; they may be stranded from an earlier enable session (deletable only via AWS support)`,
982	),
983	),
984	);
985
986	const updateTableContributorInsights = (
987	tableName: string,
988	enabled: boolean,
989	) =>
990	dynamodb
991	.updateContributorInsights({
992	TableName: tableName,
993	ContributorInsightsAction: enabled ? "ENABLE" : "DISABLE",
994	})
995	.pipe(
996	Effect.retry({
997	while: isRetryableControlPlaneError,
998	schedule: Schedule.max([
999	Schedule.exponential(250),
1000	Schedule.recurs(15),
1001	]),
1002	}),
1003	);
1004
1005	const waitForTableActive = (
1006	session: {
1007	note: (message: string) => Effect.Effect<void, never, never>;
1008	},
1009	tableName: string,
1010	) => {
1011	let elapsedSeconds = 0;
1012	let progressMessage = `DynamoDB Table provider: waiting for table ${tableName} to become ACTIVE`;
1013
1014	return Effect.gen(function* () {
1015	const response = yield* dynamodb.describeTable({
1016	TableName: tableName,
1017	});
1018	if (response.Table?.TableStatus !== "ACTIVE") {
1019	progressMessage = `DynamoDB Table provider: table ${tableName} not active yet (status=${response.Table?.TableStatus ?? "undefined"} gsiStatuses=${formatGlobalSecondaryIndexStatuses(response.Table?.GlobalSecondaryIndexes)})`;
1020	return yield* Effect.fail(new TableNotActive());
1021	}
1022	yield* session.note(
1023	`DynamoDB Table provider: table ${tableName} is ACTIVE (${formatPollingElapsed(elapsedSeconds)})`,
1024	);
1025	return response.Table;
1026	}).pipe(
1027	Effect.retry({
1028	while: (error) =>
1029	error._tag === "TableNotActive" ||
1030	isRetryableControlPlaneError(error),
1031	schedule: waitForTableActivationConvergence.pipe(
1032	Schedule.tap(({ attempt }) => {
1033	elapsedSeconds = attempt * 10;
1034	return session.note(
1035	`${progressMessage} (${formatPollingElapsed(elapsedSeconds)})`,
1036	);
1037	}),
1038	),
1039	}),
1040	);
1041	};
1042
1043	const waitForGlobalSecondaryIndexesStable = (
1044	session: {
1045	note: (message: string) => Effect.Effect<void, never, never>;
1046	},
1047	tableName: string,
1048	expectedIndexNames: readonly string[],
1049	) => {
1050	let elapsedSeconds = 0;
1051	let progressMessage = `DynamoDB Table provider: waiting for GSIs on ${tableName} to stabilize`;
1052
1053	return Effect.gen(function* () {
1054	const response = yield* dynamodb.describeTable({
1055	TableName: tableName,
1056	});
1057	const table = response.Table;
1058	const actualIndexNames = [...(table?.GlobalSecondaryIndexes ?? [])]
1059	.map((index) => index.IndexName!)
1060	.sort();
1061	const expected = [...expectedIndexNames].sort();
1062	const allActive = (table?.GlobalSecondaryIndexes ?? []).every(
1063	(index) => index.IndexStatus === "ACTIVE",
1064	);
1065
1066	if (
1067	JSON.stringify(actualIndexNames) !== JSON.stringify(expected) ||
1068	!allActive
1069	) {
1070	progressMessage = `DynamoDB Table provider: GSIs for ${tableName} not stable yet (expected=${JSON.stringify(expected)} actual=${JSON.stringify(actualIndexNames)} statuses=${JSON.stringify((table?.GlobalSecondaryIndexes ?? []).map((index) => ({ name: index.IndexName, status: index.IndexStatus })))} tableStatus=${table?.TableStatus ?? "undefined"})`;
1071	return yield* Effect.fail(new TableIndexesNotStable());
1072	}
1073
1074	yield* session.note(
1075	`DynamoDB Table provider: GSIs for ${tableName} stabilized (${JSON.stringify(actualIndexNames)}) (${formatPollingElapsed(elapsedSeconds)})`,
1076	);
1077	return table;
1078	}).pipe(
1079	Effect.retry({
1080	while: (error) =>
1081	error._tag === "TableIndexesNotStable" ||
1082	isRetryableControlPlaneError(error),
1083	schedule: waitForGlobalSecondaryIndexesConvergence.pipe(
1084	Schedule.tap(({ attempt }) => {
1085	elapsedSeconds = attempt * 10;
1086	return session.note(
1087	`${progressMessage} (${formatPollingElapsed(elapsedSeconds)})`,
1088	);
1089	}),
1090	),
1091	}),
1092	);
1093	};
1094
1095	const waitForTableDeleted = (
1096	session: {
1097	note: (message: string) => Effect.Effect<void, never, never>;
1098	},
1099	tableName: string,
1100	) => {
1101	let elapsedSeconds = 0;
1102	let progressMessage = `DynamoDB Table provider: waiting for deletion of ${tableName}`;
1103
1104	return Effect.gen(function* () {
1105	const response = yield* dynamodb.describeTable({
1106	TableName: tableName,
1107	});
1108	progressMessage = `DynamoDB Table provider: table ${tableName} still deleting (status=${response.Table?.TableStatus ?? "undefined"})`;
1109	return yield* Effect.fail(new TableStillDeleting());
1110	}).pipe(
1111	Effect.catchTag("ResourceNotFoundException", () => {
1112	return session.note(
1113	`DynamoDB Table provider: table ${tableName} deletion confirmed (${formatPollingElapsed(elapsedSeconds)})`,
1114	);
1115	}),
1116	Effect.retry({
1117	while: (error) =>
1118	error._tag === "TableStillDeleting" ||
1119	isRetryableControlPlaneError(error),
1120	schedule: waitForDeletionConvergence.pipe(
1121	Schedule.tap(({ attempt }) => {
1122	elapsedSeconds = attempt;
1123	return session.note(
1124	`${progressMessage} (${formatPollingElapsed(elapsedSeconds)})`,
1125	);
1126	}),
1127	),
1128	}),
1129	);
1130	};
1131
1132	const deleteGlobalSecondaryIndexes = (
1133	session: {
1134	note: (message: string) => Effect.Effect<void, never, never>;
1135	},
1136	tableName: string,
1137	) =>
1138	Effect.gen(function* () {
1139	const response = yield* dynamodb
1140	.describeTable({
1141	TableName: tableName,
1142	})
1143	.pipe(
1144	Effect.retry({
1145	while: isRetryableControlPlaneError,
1146	schedule: waitForControlPlaneConvergence,
1147	}),
1148	Effect.catchTag("ResourceNotFoundException", () =>
1149	Effect.succeed({ Table: undefined }),
1150	),
1151	);
1152
1153	const indexNames = [...(response.Table?.GlobalSecondaryIndexes ?? [])]
1154	.map((index) => index.IndexName!)
1155	.sort();
1156
1157	if (indexNames.length === 0) {
1158	yield* session.note(
1159	`DynamoDB Table provider: no GSIs remain on ${tableName}; retrying table deletion`,
1160	);
1161	return;
1162	}
1163
1164	yield* session.note(
1165	`DynamoDB Table provider: deleting GSIs before deleting table ${tableName} (${indexNames.join(", ")})`,
1166	);
1167
1168	yield* waitForGlobalSecondaryIndexesStable(
1169	session,
1170	tableName,
1171	indexNames,
1172	);
1173
1174	const remainingIndexNames = [...indexNames];
1175	for (const indexName of indexNames) {
1176	let elapsedSeconds = 0;
1177	let progressMessage = `DynamoDB Table provider: waiting to delete GSI ${indexName} from ${tableName}`;
1178
1179	yield* session.note(
1180	`DynamoDB Table provider: deleting GSI ${indexName} from ${tableName}`,
1181	);
1182
1183	yield* dynamodb
1184	.updateTable({
1185	TableName: tableName,
1186	GlobalSecondaryIndexUpdates: [
1187	{
1188	Delete: {
1189	IndexName: indexName,
1190	},
1191	},
1192	],
1193	})
1194	.pipe(
1195	Effect.timeout(1000),
1196	Effect.tap(() =>
1197	session.note(
1198	`DynamoDB Table provider: delete accepted for GSI ${indexName} on ${tableName} (${formatPollingElapsed(elapsedSeconds)})`,
1199	),
1200	),
1201	Effect.retry({
1202	while: (error) => {
1203	if (error._tag === "ResourceInUseException") {
1204	progressMessage = `DynamoDB Table provider: delete for GSI ${indexName} on ${tableName} is blocked while the table or indexes are still transitioning`;
1205	return true;
1206	}
1207	if (error._tag === "TimeoutError") {
1208	progressMessage = `DynamoDB Table provider: delete for GSI ${indexName} on ${tableName} timed out`;
1209	return true;
1210	}
1211	if (
1212	error._tag === "InternalServerError" ||
1213	error._tag === "LimitExceededException"
1214	) {
1215	progressMessage = `DynamoDB Table provider: delete for GSI ${indexName} on ${tableName} hit ${error._tag}`;
1216	return true;
1217	}
1218	return false;
1219	},
1220	schedule: waitForGlobalSecondaryIndexesConvergence.pipe(
1221	Schedule.tap(({ attempt }) => {
1222	elapsedSeconds = attempt * 10;
1223	return session.note(
1224	`${progressMessage} (${formatPollingElapsed(elapsedSeconds)})`,
1225	);
1226	}),
1227	),
1228	}),
1229	);
1230
1231	remainingIndexNames.splice(
1232	remainingIndexNames.indexOf(indexName),
1233	1,
1234	);
1235
1236	yield* waitForGlobalSecondaryIndexesStable(
1237	session,
1238	tableName,
1239	remainingIndexNames,
1240	);
1241	}
1242	});
1243
1244	// For read, we want `ResourceNotFoundException` to propagate immediately
1245	// (the table doesn't exist — this is the cold-start adoption "no
1246	// resource" signal). Only retry on transient control-plane errors.
1247	const isRetryableReadError = (error: { _tag?: string }) =>
1248	error._tag === "InternalServerError" ||
1249	error._tag === "LimitExceededException" ||
1250	// See `isRetryableControlPlaneError`: read-path describes throttle too
1251	// under full-suite concurrency, so ride them out generously.
1252	error._tag === "ThrottlingException" ||
1253	error._tag === "ProvisionedThroughputExceededException" ||
1254	error._tag === "RequestLimitExceeded";
1255
1256	const readTableState = (tableName: string) =>
1257	Effect.gen(function* () {
1258	const response = yield* dynamodb
1259	.describeTable({
1260	TableName: tableName,
1261	})
1262	.pipe(
1263	Effect.retry({
1264	while: isRetryableReadError,
1265	schedule: Schedule.max([
1266	Schedule.exponential(250),
1267	Schedule.recurs(30),
1268	]),
1269	}),
1270	);
1271	const table = response.Table;
1272	if (!table?.TableArn) {
1273	return yield* Effect.fail(
1274	new Error(`Table ${tableName} not found`),
1275	);
1276	}
1277
1278	const [tagsResult, continuousBackupsResult, ttlResult] =
1279	yield* Effect.all([
1280	dynamodb
1281	.listTagsOfResource({
1282	ResourceArn: table.TableArn,
1283	})
1284	.pipe(
1285	Effect.retry({
1286	while: isRetryableReadError,
1287	schedule: Schedule.max([
1288	Schedule.exponential(250),
1289	Schedule.recurs(30),
1290	]),
1291	}),
1292	),
1293	dynamodb
1294	.describeContinuousBackups({
1295	TableName: tableName,
1296	})
1297	.pipe(
1298	Effect.retry({
1299	while: (e) => e._tag === "InternalServerError",
1300	schedule: Schedule.max([
1301	Schedule.exponential(250),
1302	Schedule.recurs(30),
1303	]),
1304	}),
1305	Effect.catchTag("TableNotFoundException", () =>
1306	Effect.succeed({ ContinuousBackupsDescription: undefined }),
1307	),
1308	),
1309	dynamodb
1310	.describeTimeToLive({
1311	TableName: tableName,
1312	})
1313	.pipe(
1314	Effect.retry({
1315	while: isRetryableReadError,
1316	schedule: Schedule.max([
1317	Schedule.exponential(250),
1318	Schedule.recurs(30),
1319	]),
1320	}),
1321	Effect.catchTag("ResourceNotFoundException", () =>
1322	Effect.succeed({ TimeToLiveDescription: undefined }),
1323	),
1324	),
1325	]);
1326
1327	return {
1328	table,
1329	tags: Object.fromEntries(
1330	(tagsResult.Tags ?? []).map((tag) => [tag.Key!, tag.Value!]),
1331	) as Record<string, string>,
1332	pointInTimeRecoveryDescription:
1333	continuousBackupsResult.ContinuousBackupsDescription
1334	?.PointInTimeRecoveryDescription,
1335	timeToLiveDescription: ttlResult.TimeToLiveDescription,
1336	};
1337	}).pipe(
1338	Effect.catchTag("ResourceNotFoundException", () =>
1339	Effect.succeed(undefined),
1340	),
1341	);
1342
1343	const toAttrs = (state: {
1344	table: DynamoDB.TableDescription;
1345	tags: Record<string, string>;
1346	pointInTimeRecoveryDescription:
1347	| DynamoDB.PointInTimeRecoveryDescription
1348	| undefined;
1349	}) => ({
1350	tableId: state.table.TableId!,
1351	tableName: state.table.TableName!,
1352	tableArn: state.table.TableArn! as TableArn,
1353	partitionKey:
1354	state.table.KeySchema?.find((key) => key.KeyType === "HASH")
1355	?.AttributeName ?? "",
1356	sortKey: state.table.KeySchema?.find((key) => key.KeyType === "RANGE")
1357	?.AttributeName,
1358	latestStreamArn: state.table.LatestStreamArn,
1359	streamSpecification: state.table.StreamSpecification,
1360	localSecondaryIndexes: state.table.LocalSecondaryIndexes,
1361	globalSecondaryIndexes: state.table.GlobalSecondaryIndexes,
1362	pointInTimeRecoveryDescription: state.pointInTimeRecoveryDescription,
1363	tags: state.tags,
1364	});
1365
1366	const indexesByName = <T extends { IndexName?: string }>(
1367	indexes: readonly T[] | undefined,
1368	) =>
1369	Object.fromEntries(
1370	(indexes ?? []).map((index) => [index.IndexName!, index]),
1371	) as Record<string, T>;
1372
1373	// KeySchema order is significant with multi-attribute keys: partition
1374	// attributes are hashed together in declaration order and sort key
1375	// attributes are queried left-to-right. Only the HASH/RANGE grouping is
1376	// normalized (DescribeTable reports all HASH elements before all RANGE
1377	// elements regardless of how the request interleaved them); the relative
1378	// order within each group must be preserved — reordering attributes
1379	// defines a different index.
1380	const normalizeKeySchema = (
1381	keySchema: readonly DynamoDB.KeySchemaElement[] | undefined,
1382	) => {
1383	const elements = (keySchema ?? []).map((element) => ({
1384	AttributeName: element.AttributeName,
1385	KeyType: element.KeyType,
1386	}));
1387	return [
1388	...elements.filter((element) => element.KeyType === "HASH"),
1389	...elements.filter((element) => element.KeyType === "RANGE"),
1390	];
1391	};
1392
1393	const normalizeProjection = (
1394	projection: DynamoDB.Projection | undefined,
1395	) => ({
1396	...projection,
1397	NonKeyAttributes: [...(projection?.NonKeyAttributes ?? [])].sort(),
1398	});
1399
1400	const isSameGsiDefinition = (
1401	left: DynamoDB.GlobalSecondaryIndex,
1402	right: DynamoDB.GlobalSecondaryIndex,
1403	) =>
1404	JSON.stringify(normalizeKeySchema(left.KeySchema)) ===
1405	JSON.stringify(normalizeKeySchema(right.KeySchema)) &&
1406	JSON.stringify(normalizeProjection(left.Projection)) ===
1407	JSON.stringify(normalizeProjection(right.Projection));
1408
1409	// The describe response includes server-side bookkeeping fields
1410	// (`NumberOfDecreasesToday`, `LastIncreaseDateTime`, etc.) that the
1411	// request shape doesn't. For PAY_PER_REQUEST tables it also reports
1412	// a zero-throughput object. Normalize both sides to the user-settable
1413	// fields so we don't fire a phantom UpdateGlobalSecondaryIndex with
1414	// `ProvisionedThroughput: undefined` (which AWS rejects with
1415	// ValidationException).
1416	const normalizeProvisioned = (
1417	pt:
1418	| { ReadCapacityUnits?: number; WriteCapacityUnits?: number }
1419	| undefined,
1420	) => {
1421	if (
1422	!pt ||
1423	((pt.ReadCapacityUnits ?? 0) === 0 &&
1424	(pt.WriteCapacityUnits ?? 0) === 0)
1425	) {
1426	return undefined;
1427	}
1428	return {
1429	ReadCapacityUnits: pt.ReadCapacityUnits,
1430	WriteCapacityUnits: pt.WriteCapacityUnits,
1431	};
1432	};
1433
1434	const normalizeOnDemand = (
1435	od:
1436	| {
1437	MaxReadRequestUnits?: number;
1438	MaxWriteRequestUnits?: number;
1439	}
1440	| undefined,
1441	) => {
1442	if (
1443	!od ||
1444	((od.MaxReadRequestUnits ?? -1) < 0 &&
1445	(od.MaxWriteRequestUnits ?? -1) < 0)
1446	) {
1447	return undefined;
1448	}
1449	return {
1450	MaxReadRequestUnits: od.MaxReadRequestUnits,
1451	MaxWriteRequestUnits: od.MaxWriteRequestUnits,
1452	};
1453	};
1454
1455	const normalizeWarm = (
1456	wt:
1457	| { ReadUnitsPerSecond?: number; WriteUnitsPerSecond?: number }
1458	| undefined,
1459	) => {
1460	if (
1461	!wt ||
1462	((wt.ReadUnitsPerSecond ?? 0) === 0 &&
1463	(wt.WriteUnitsPerSecond ?? 0) === 0)
1464	) {
1465	return undefined;
1466	}
1467	return {
1468	ReadUnitsPerSecond: wt.ReadUnitsPerSecond,
1469	WriteUnitsPerSecond: wt.WriteUnitsPerSecond,
1470	};
1471	};
1472
1473	const diffGlobalSecondaryIndexes = (
1474	olds: readonly DynamoDB.GlobalSecondaryIndex[] | undefined,
1475	news: readonly DynamoDB.GlobalSecondaryIndex[] | undefined,
1476	) => {
1477	const oldByName = indexesByName(olds);
1478	const newByName = indexesByName(news);
1479	const updates: DynamoDB.GlobalSecondaryIndexUpdate[] = [];
1480	let requiresReplacement = false;
1481
1482	for (const [indexName, oldIndex] of Object.entries(oldByName)) {
1483	const newIndex = newByName[indexName];
1484	if (!newIndex) {
1485	updates.push({
1486	Delete: {
1487	IndexName: indexName,
1488	},
1489	});
1490	continue;
1491	}
1492
1493	if (!isSameGsiDefinition(oldIndex, newIndex)) {
1494	requiresReplacement = true;
1495	continue;
1496	}
1497
1498	// Only emit an Update when the user has explicitly set a
1499	// throughput value AND it differs from observed. Cloud-side
1500	// defaults (AWS auto-assigns ProvisionedThroughput/WarmThroughput
1501	// values for PAY_PER_REQUEST tables) should not trigger phantom
1502	// updates against an undefined desired value.
1503	const provDiff =
1504	newIndex.ProvisionedThroughput !== undefined &&
1505	JSON.stringify(
1506	normalizeProvisioned(oldIndex.ProvisionedThroughput),
1507	) !==
1508	JSON.stringify(
1509	normalizeProvisioned(newIndex.ProvisionedThroughput),
1510	);
1511	const odDiff =
1512	newIndex.OnDemandThroughput !== undefined &&
1513	JSON.stringify(normalizeOnDemand(oldIndex.OnDemandThroughput)) !==
1514	JSON.stringify(normalizeOnDemand(newIndex.OnDemandThroughput));
1515	const wtDiff =
1516	newIndex.WarmThroughput !== undefined &&
1517	JSON.stringify(normalizeWarm(oldIndex.WarmThroughput)) !==
1518	JSON.stringify(normalizeWarm(newIndex.WarmThroughput));
1519	if (provDiff || odDiff || wtDiff) {
1520	updates.push({
1521	Update: {
1522	IndexName: indexName,
1523	ProvisionedThroughput: newIndex.ProvisionedThroughput,
1524	OnDemandThroughput: newIndex.OnDemandThroughput,
1525	WarmThroughput: newIndex.WarmThroughput,
1526	},
1527	});
1528	}
1529	}
1530
1531	for (const [indexName, newIndex] of Object.entries(newByName)) {
1532	if (!oldByName[indexName]) {
1533	updates.push({
1534	Create: {
1535	IndexName: indexName,
1536	KeySchema: newIndex.KeySchema,
1537	Projection: newIndex.Projection,
1538	ProvisionedThroughput: newIndex.ProvisionedThroughput,
1539	OnDemandThroughput: newIndex.OnDemandThroughput,
1540	WarmThroughput: newIndex.WarmThroughput,
1541	},
1542	});
1543	}
1544	}
1545
1546	return {
1547	updates,
1548	requiresReplacement,
1549	};
1550	};
1551
1552	return Table.Provider.of({
1553	stables: ["tableName", "tableId", "tableArn"],
1554	// Enumerate every table in the ambient account/region. `listTables`
1555	// returns only names, so each is hydrated to the full Attributes shape
1556	// via the same multi-API read helper (`readTableState`) `read` uses.
1557	list: () =>
1558	Effect.gen(function* () {
1559	const names = yield* dynamodb.listTables.items({}).pipe(
1560	Stream.runCollect,
1561	Effect.map((chunk) => Array.from(chunk)),
1562	);
1563	const states = yield* Effect.forEach(
1564	names,
1565	(tableName) =>
1566	readTableState(tableName).pipe(
1567	// Hydrating every table fires four describe calls each.
1568	// Across a busy account this trips DynamoDB's read throttle,
1569	// and a table that isn't ACTIVE yet (a peer mid-create)
1570	// transiently rejects the continuous-backups/TTL describes
1571	// with ValidationException. Both are transient — retry to
1572	// converge so a table that *is* ours still hydrates fully.
1573	Effect.retry({
1574	while: (e) =>
1575	e._tag === "ThrottlingException" ||
1576	e._tag === "ValidationException",
1577	schedule: Schedule.max([
1578	Schedule.exponential(250).pipe(Schedule.jittered),
1579	Schedule.recurs(12),
1580	]),
1581	}),
1582	// Last resort: if a foreign table never settles within the
1583	// retry budget (e.g. a peer is still mid-create/delete when
1584	// we give up), skip it rather than failing the whole
1585	// enumeration. Our own table is ACTIVE by the time list()
1586	// runs, so it always hydrates via the retry above.
1587	//
1588	// `AccessDeniedException` is the same case seen from the
1589	// other side: enumerating an account walks tables we do not
1590	// own, and one can deny the tag read outright (a restrictive
1591	// resource policy live; a peer test's table under the local
1592	// emulator). A table we cannot read is not ours to return.
1593	// `TableNotFoundException` completes the set: a peer can
1594	// delete a table between `listTables` and our describes, and
1595	// that is what DynamoDB raises for a table that vanished.
1596	Effect.catchTag(
1597	[
1598	"ValidationException",
1599	"AccessDeniedException",
1600	"TableNotFoundException",
1601	"ResourceNotFoundException",
1602	],
1603	() => Effect.succeed(undefined),
1604	),
1605	),
1606	{ concurrency: 8 },
1607	);
1608	return states
1609	.filter((state) => state !== undefined)
1610	.map((state) => toAttrs(state));
1611	}),
1612	read: Effect.fn(function* ({ id, olds, output }) {
1613	const tableName =
1614	output?.tableName ??
1615	(olds ? yield* createTableName(id, olds) : undefined);
1616	if (!tableName) return undefined;
1617	const state = yield* readTableState(tableName).pipe(
1618	Effect.catchTag("ResourceNotFoundException", () =>
1619	Effect.succeed(undefined),
1620	),
1621	);
1622	if (!state) return undefined;
1623	const attrs = toAttrs(state);
1624	return (yield* hasAlchemyTags(id, state.tags as any))
1625	? attrs
1626	: Unowned(attrs);
1627	}),
1628	diff: Effect.fn(function* ({ news, olds }) {
1629	if (!isResolved(news)) return undefined;
1630	// Renaming the table always replaces. The successor's name cannot
1631	// collide with the predecessor's, so the default
1632	// create-before-delete replacement flow applies.
1633	if (news.tableName !== olds.tableName) {
1634	return { action: "replace" } as const;
1635	}
1636	// For every other replacement trigger the physical name stays the
1637	// same. Engine-generated names embed the instance id, which is
1638	// re-minted on replacement, so the successor gets a fresh name and
1639	// create-before-delete works. A hard-coded `tableName` cannot
1640	// change — DynamoDB table names are unique per account/region, so
1641	// the successor's createTable would collide with the still-live
1642	// predecessor and mis-adopt it. Delete the old generation first in
1643	// that case; `delete` waits until the name is fully released.
1644	const replace =
1645	news.tableName !== undefined
1646	? ({ action: "replace", deleteFirst: true } as const)
1647	: ({ action: "replace" } as const);
1648	if (
1649	olds.partitionKey !== news.partitionKey ||
1650	olds.sortKey !== news.sortKey
1651	) {
1652	return replace;
1653	}
1654	for (const [name, type] of Object.entries(olds.attributes ?? {})) {
1655	if (news.attributes[name] !== type) {
1656	return replace;
1657	}
1658	}
1659	// Compare secondary indexes in the wire shape so legacy state
1660	// (persisted before the typed index props) diffs cleanly against
1661	// the new prop shape.
1662	if (
1663	havePropsChanged(
1664	{
1665	localSecondaryIndexes:
1666	toWireLocalSecondaryIndexes(
1667	olds.partitionKey,
1668	olds.localSecondaryIndexes,
1669	) ?? [],
1670	},
1671	{
1672	localSecondaryIndexes:
1673	toWireLocalSecondaryIndexes(
1674	news.partitionKey,
1675	news.localSecondaryIndexes,
1676	) ?? [],
1677	},
1678	)
1679	) {
1680	return replace;
1681	}
1682	const { requiresReplacement } = diffGlobalSecondaryIndexes(
1683	toWireGlobalSecondaryIndexes(olds.globalSecondaryIndexes),
1684	toWireGlobalSecondaryIndexes(news.globalSecondaryIndexes),
1685	);
1686	if (requiresReplacement) {
1687	return replace;
1688	}
1689	// ImportSourceSpecification (importTable) is out of scope — see
1690	// processes/AWS/catalog/serverless-data.md (import/export are
1691	// long-running jobs, deferred as effectful functions).
1692	}),
1693
1694	reconcile: Effect.fn(function* ({
1695	id,
1696	output,
1697	news,
1698	session,
1699	bindings,
1700	}) {
1701	const tableName =
1702	output?.tableName ?? (yield* createTableName(id, news));
1703	const desiredTags = yield* createTags(id, news.tags);
1704	const desiredStreamSpecification =
1705	yield* resolveStreamSpecification(bindings);
1706
1707	// Observe cloud state. `output` is treated as a cache for the table
1708	// name; the table's actual existence and configuration are fetched
1709	// fresh so the reconciler converges regardless of drift, adoption,
1710	// or a partially-completed prior run.
1711	let state = yield* readTableState(tableName);
1712
1713	if (state === undefined) {
1714	yield* session.note(
1715	`Table ${tableName}: creating with stream specification ${JSON.stringify(desiredStreamSpecification)}`,
1716	);
1717	yield* dynamodb
1718	.createTable({
1719	TableName: tableName,
1720	TableClass: news.tableClass,
1721	KeySchema: toKeySchema(news),
1722	AttributeDefinitions: toAttributeDefinitions(news.attributes),
1723	LocalSecondaryIndexes: toWireLocalSecondaryIndexes(
1724	news.partitionKey,
1725	news.localSecondaryIndexes,
1726	),
1727	GlobalSecondaryIndexes: toWireGlobalSecondaryIndexes(
1728	news.globalSecondaryIndexes,
1729	),
1730	BillingMode: news.billingMode ?? "PAY_PER_REQUEST",
1731	SSESpecification: news.sseSpecification,
1732	StreamSpecification: desiredStreamSpecification,
1733	WarmThroughput: news.warmThroughput,
1734	DeletionProtectionEnabled: news.deletionProtectionEnabled,
1735	OnDemandThroughput: news.onDemandThroughput,
1736	ProvisionedThroughput: news.provisionedThroughput,
1737	Tags: createTagsList(desiredTags),
1738	})
1739	.pipe(
1740	Effect.retry({
1741	while: (e) =>
1742	e._tag === "LimitExceededException" ||
1743	e._tag === "InternalServerError",
1744	schedule: Schedule.exponential(100),
1745	}),
1746	// A peer reconciler created the table between our observe and
1747	// create; describe it and continue with the sync path.
1748	Effect.catchTag("ResourceInUseException", () =>
1749	adoptExistingTable(tableName),
1750	),
1751	);
1752
1753	yield* waitForTableActive(session, tableName);
1754
1755	if ((news.globalSecondaryIndexes?.length ?? 0) > 0) {
1756	yield* waitForGlobalSecondaryIndexesStable(
1757	session,
1758	tableName,
1759	news.globalSecondaryIndexes?.map((index) => index.indexName) ??
1760	[],
1761	);
1762	}
1763
1764	state = yield* readTableState(tableName);
1765	if (!state) {
1766	return yield* Effect.fail(
1767	new Error(`Failed to read created table ${tableName}`),
1768	);
1769	}
1770	}
1771
1772	// Sync stream specification — observed ↔ desired.
1773	//
1774	// SQL of stream changes: changing the StreamViewType requires a
1775	// disable→enable sequence; AWS rejects an in-place view-type change.
1776	const currentStreamSpecification = normalizeStreamSpecification(
1777	state.table.StreamSpecification,
1778	);
1779	const streamViewTypeChanged =
1780	currentStreamSpecification?.StreamEnabled === true &&
1781	desiredStreamSpecification?.StreamEnabled === true &&
1782	currentStreamSpecification.StreamViewType !==
1783	desiredStreamSpecification.StreamViewType;
1784
1785	if (streamViewTypeChanged) {
1786	yield* dynamodb.updateTable({
1787	TableName: tableName,
1788	StreamSpecification: { StreamEnabled: false },
1789	});
1790	yield* waitForTableActive(session, tableName);
1791	}
1792
1793	if (
1794	havePropsChanged(
1795	{ streamSpecification: currentStreamSpecification },
1796	{ streamSpecification: desiredStreamSpecification },
1797	)
1798	) {
1799	yield* session.note(
1800	`Table ${tableName}: updating stream configuration`,
1801	);
1802	yield* dynamodb.updateTable({
1803	TableName: tableName,
1804	StreamSpecification: desiredStreamSpecification ?? {
1805	StreamEnabled: false,
1806	},
1807	});
1808	yield* waitForTableActive(session, tableName);
1809	}
1810
1811	// Sync GSIs — diff observed cloud state against desired and apply
1812	// each delta serially. AWS only accepts one GSI change per
1813	// updateTable call. The diff function only reads fields shared
1814	// between `GlobalSecondaryIndex` and `GlobalSecondaryIndexDescription`
1815	// (IndexName, KeySchema, Projection, throughputs), so the cast is safe.
1816	const { updates: globalSecondaryIndexUpdates } =
1817	diffGlobalSecondaryIndexes(
1818	state.table.GlobalSecondaryIndexes as
1819	| readonly DynamoDB.GlobalSecondaryIndex[]
1820	| undefined,
1821	toWireGlobalSecondaryIndexes(news.globalSecondaryIndexes),
1822	);
1823
1824	for (const globalSecondaryIndexUpdate of globalSecondaryIndexUpdates) {
1825	const action = globalSecondaryIndexUpdate.Create
1826	? `create ${globalSecondaryIndexUpdate.Create.IndexName}`
1827	: globalSecondaryIndexUpdate.Update
1828	? `update ${globalSecondaryIndexUpdate.Update.IndexName}`
1829	: `delete ${globalSecondaryIndexUpdate.Delete!.IndexName}`;
1830	yield* session.note(
1831	`Table ${tableName}: applying GSI update (${action})`,
1832	);
1833	yield* dynamodb.updateTable({
1834	TableName: tableName,
1835	AttributeDefinitions: toAttributeDefinitions(news.attributes),
1836	GlobalSecondaryIndexUpdates: [globalSecondaryIndexUpdate],
1837	});
1838	yield* waitForTableActive(session, tableName);
1839	}
1840
1841	if (globalSecondaryIndexUpdates.length > 0) {
1842	const expectedNames =
1843	news.globalSecondaryIndexes?.map((index) => index.indexName) ??
1844	[];
1845	yield* session.note(
1846	`Table ${tableName}: waiting for GSIs to stabilize (${expectedNames.join(", ") || "none"})`,
1847	);
1848	yield* waitForGlobalSecondaryIndexesStable(
1849	session,
1850	tableName,
1851	expectedNames,
1852	);
1853	yield* session.note(`Table ${tableName}: GSIs stabilized`);
1854	}
1855
1856	// Sync base attributes — observed ↔ desired.
1857	const desiredBillingMode = news.billingMode ?? "PAY_PER_REQUEST";
1858	// DynamoDB omits `BillingModeSummary` for tables that have always
1859	// been PROVISIONED — fall back to the observed throughput (on-demand
1860	// tables report 0 read capacity) so an unchanged PROVISIONED table
1861	// does not trigger a spurious `updateTable` (which DynamoDB rejects
1862	// with "provisioned throughput will not change").
1863	const observedBillingMode =
1864	state.table.BillingModeSummary?.BillingMode ??
1865	((state.table.ProvisionedThroughput?.ReadCapacityUnits ?? 0) > 0
1866	? "PROVISIONED"
1867	: "PAY_PER_REQUEST");
1868	const observedProvisionedThroughput = state.table
1869	.ProvisionedThroughput
1870	? {
1871	ReadCapacityUnits:
1872	state.table.ProvisionedThroughput.ReadCapacityUnits,
1873	WriteCapacityUnits:
1874	state.table.ProvisionedThroughput.WriteCapacityUnits,
1875	}
1876	: undefined;
1877	const baseChanged = havePropsChanged(
1878	{
1879	tableClass: state.table.TableClassSummary?.TableClass,
1880	billingMode: observedBillingMode,
1881	deletionProtectionEnabled:
1882	state.table.DeletionProtectionEnabled ?? false,
1883	provisionedThroughput:
1884	desiredBillingMode === "PROVISIONED"
1885	? observedProvisionedThroughput
1886	: undefined,
1887	},
1888	{
1889	tableClass: news.tableClass,
1890	billingMode: desiredBillingMode,
1891	deletionProtectionEnabled:
1892	news.deletionProtectionEnabled ?? false,
1893	provisionedThroughput:
1894	desiredBillingMode === "PROVISIONED"
1895	? news.provisionedThroughput
1896	: undefined,
1897	},
1898	);
1899
1900	if (baseChanged) {
1901	yield* dynamodb.updateTable({
1902	TableName: tableName,
1903	TableClass: news.tableClass,
1904	AttributeDefinitions: toAttributeDefinitions(news.attributes),
1905	BillingMode: desiredBillingMode,
1906	SSESpecification: news.sseSpecification,
1907	WarmThroughput: news.warmThroughput,
1908	DeletionProtectionEnabled: news.deletionProtectionEnabled,
1909	OnDemandThroughput: news.onDemandThroughput,
1910	ProvisionedThroughput: news.provisionedThroughput,
1911	});
1912	yield* waitForTableActive(session, tableName);
1913	}
1914
1915	// Sync TTL — observed ↔ desired.
1916	//
1917	// updateTimeToLive rejects calls that would not change anything
1918	// (e.g. enabling TTL on the same attribute name), so we have to
1919	// diff against the current TimeToLiveDescription before calling.
1920	const currentTtl = state.timeToLiveDescription;
1921	const currentTtlEnabled =
1922	currentTtl?.TimeToLiveStatus === "ENABLED" ||
1923	currentTtl?.TimeToLiveStatus === "ENABLING";
1924	const currentTtlAttribute = currentTtl?.AttributeName;
1925	const desiredTtl = news.timeToLiveSpecification;
1926	const desiredTtlEnabled = desiredTtl?.Enabled === true;
1927	const desiredTtlAttribute = desiredTtl?.AttributeName;
1928
1929	if (
1930	desiredTtlEnabled !== currentTtlEnabled ||
1931	(desiredTtlEnabled && desiredTtlAttribute !== currentTtlAttribute)
1932	) {
1933	if (desiredTtlEnabled) {
1934	if (
1935	currentTtlEnabled &&
1936	currentTtlAttribute !== desiredTtlAttribute
1937	) {
1938	// AWS only allows one TTL attribute. We must disable first
1939	// before re-enabling on a different attribute.
1940	yield* updateTimeToLive(tableName, {
1941	AttributeName: currentTtlAttribute!,
1942	Enabled: false,
1943	});
1944	}
1945	yield* updateTimeToLive(tableName, {
1946	AttributeName: desiredTtlAttribute!,
1947	Enabled: true,
1948	});
1949	} else if (currentTtlEnabled && currentTtlAttribute) {
1950	yield* updateTimeToLive(tableName, {
1951	AttributeName: currentTtlAttribute,
1952	Enabled: false,
1953	});
1954	}
1955	}
1956
1957	// Sync PITR — observed ↔ desired.
1958	const currentPitrEnabled =
1959	state.pointInTimeRecoveryDescription?.PointInTimeRecoveryStatus ===
1960	"ENABLED";
1961	const desiredPitrEnabled =
1962	news.pointInTimeRecoverySpecification?.pointInTimeRecoveryEnabled ??
1963	false;
1964	const currentPitrPeriod =
1965	state.pointInTimeRecoveryDescription?.RecoveryPeriodInDays;
1966	const desiredPitrPeriod = toWireDays(
1967	news.pointInTimeRecoverySpecification?.recoveryPeriod,
1968	);
1969	if (
1970	currentPitrEnabled !== desiredPitrEnabled ||
1971	(desiredPitrEnabled && currentPitrPeriod !== desiredPitrPeriod)
1972	) {
1973	yield* updateContinuousBackups(tableName, {
1974	PointInTimeRecoveryEnabled: desiredPitrEnabled,
1975	RecoveryPeriodInDays: desiredPitrPeriod,
1976	});
1977	}
1978
1979	// Sync resource policy — observed ↔ desired. The observed policy is
1980	// read fresh from the cloud (never olds/output) so adoption and
1981	// out-of-band drift converge.
1982	const observedPolicy = yield* readTableResourcePolicy(
1983	state.table.TableArn!,
1984	);
1985	if (
1986	canonicalPolicyDocument(observedPolicy) !==
1987	canonicalPolicyDocument(news.resourcePolicy)
1988	) {
1989	if (news.resourcePolicy !== undefined) {
1990	yield* session.note(
1991	`Table ${tableName}: putting resource policy`,
1992	);
1993	yield* putTableResourcePolicy(
1994	state.table.TableArn!,
1995	news.resourcePolicy,
1996	);
1997	} else {
1998	yield* session.note(
1999	`Table ${tableName}: deleting resource policy`,
2000	);
2001	yield* deleteTableResourcePolicy(state.table.TableArn!);
2002	}
2003	}
2004
2005	// Sync Kinesis streaming destination — observe the table's actual
2006	// destinations, wait out any in-flight transition (enable/disable
2007	// reject while another change is in progress), disable whatever is
2008	// ACTIVE but not desired, then enable/update the desired
2009	// destination and wait for it to reach ACTIVE.
2010	const desiredKinesisDestination = news.kinesisStreamingDestination;
2011	let kinesisDestinations = yield* waitForKinesisDestinationsSettled(
2012	session,
2013	tableName,
2014	);
2015	for (const destination of kinesisDestinations) {
2016	if (destination.DestinationStatus !== "ACTIVE") continue;
2017	if (
2018	destination.StreamArn === desiredKinesisDestination?.streamArn
2019	) {
2020	continue;
2021	}
2022	yield* session.note(
2023	`Table ${tableName}: disabling Kinesis streaming destination ${destination.StreamArn}`,
2024	);
2025	yield* disableKinesisDestination(tableName, destination.StreamArn!);
2026	kinesisDestinations = yield* waitForKinesisDestinationsSettled(
2027	session,
2028	tableName,
2029	);
2030	}
2031	if (desiredKinesisDestination !== undefined) {
2032	const activeDestination = kinesisDestinations.find(
2033	(destination) =>
2034	destination.StreamArn === desiredKinesisDestination.streamArn &&
2035	destination.DestinationStatus === "ACTIVE",
2036	);
2037	if (activeDestination === undefined) {
2038	yield* session.note(
2039	`Table ${tableName}: enabling Kinesis streaming destination ${desiredKinesisDestination.streamArn}`,
2040	);
2041	yield* enableKinesisDestination(
2042	tableName,
2043	desiredKinesisDestination,
2044	);
2045	kinesisDestinations = yield* waitForKinesisDestinationsSettled(
2046	session,
2047	tableName,
2048	);
2049	} else if (
2050	desiredKinesisDestination.approximateCreationDateTimePrecision !==
2051	undefined &&
2052	(activeDestination.ApproximateCreationDateTimePrecision ??
2053	"MILLISECOND") !==
2054	desiredKinesisDestination.approximateCreationDateTimePrecision
2055	) {
2056	yield* session.note(
2057	`Table ${tableName}: updating Kinesis streaming destination precision to ${desiredKinesisDestination.approximateCreationDateTimePrecision}`,
2058	);
2059	yield* updateKinesisDestinationPrecision(
2060	tableName,
2061	desiredKinesisDestination,
2062	);
2063	kinesisDestinations = yield* waitForKinesisDestinationsSettled(
2064	session,
2065	tableName,
2066	);
2067	}
2068	const settledDestination = kinesisDestinations.find(
2069	(destination) =>
2070	destination.StreamArn === desiredKinesisDestination.streamArn,
2071	);
2072	if (settledDestination?.DestinationStatus !== "ACTIVE") {
2073	return yield* Effect.fail(
2074	new KinesisStreamingDestinationFailed({
2075	tableName,
2076	streamArn: desiredKinesisDestination.streamArn,
2077	status: settledDestination?.DestinationStatus,
2078	description: settledDestination?.DestinationStatusDescription,
2079	}),
2080	);
2081	}
2082	}
2083
2084	// Sync Contributor Insights — observed ↔ desired.
2085	const desiredInsightsEnabled =
2086	news.contributorInsightsEnabled ?? false;
2087	const observedInsightsStatus =
2088	yield* waitForContributorInsightsSettled(session, tableName);
2089	const observedInsightsEnabled = observedInsightsStatus === "ENABLED";
2090	if (observedInsightsEnabled !== desiredInsightsEnabled) {
2091	yield* session.note(
2092	`Table ${tableName}: ${desiredInsightsEnabled ? "enabling" : "disabling"} Contributor Insights`,
2093	);
2094	yield* updateTableContributorInsights(
2095	tableName,
2096	desiredInsightsEnabled,
2097	);
2098	if (!desiredInsightsEnabled) {
2099	// Wait for the DISABLE to settle so DynamoDB's rule cleanup
2100	// (which fires on the DISABLING→DISABLED transition) runs
2101	// while the table still exists — a deleteTable racing this
2102	// window strands the CloudWatch rules forever.
2103	yield* waitForContributorInsightsSettled(session, tableName);
2104	}
2105	}
2106
2107	// Sync tags — diff observed cloud tags against desired.
2108	//
2109	// Adoption may bring us a table that already has its own tag set;
2110	// diffing against `state.tags` (fetched fresh) lets the reconciler
2111	// converge ownership without fighting whatever was there before.
2112	const { removed, upsert } = diffTags(state.tags, desiredTags);
2113	if (removed.length > 0) {
2114	yield* dynamodb.untagResource({
2115	ResourceArn: state.table.TableArn!,
2116	TagKeys: removed,
2117	});
2118	}
2119	if (upsert.length > 0) {
2120	yield* dynamodb.tagResource({
2121	ResourceArn: state.table.TableArn!,
2122	Tags: upsert,
2123	});
2124	}
2125
2126	// Re-read final state after all sync steps so the returned
2127	// attributes reflect the post-reconcile cloud state.
2128	const final = yield* readTableState(tableName);
2129	if (!final) {
2130	return yield* Effect.fail(
2131	new Error(`Failed to read reconciled table ${tableName}`),
2132	);
2133	}
2134
2135	yield* session.note(final.table.TableArn!);
2136
2137	return {
2138	...toAttrs(final),
2139	tags: desiredTags,
2140	};
2141	}),
2142
2143	delete: Effect.fn(function* ({ output, session }) {
2144	// Contributor Insights must be fully torn down BEFORE the table is
2145	// deleted: DynamoDB removes its CloudWatch insight rules only when
2146	// a DISABLE completes while the table exists. Deleting the table
2147	// while insights are ENABLED (or mid-transition) aborts that
2148	// cleanup and strands the rules — CloudWatch rejects direct
2149	// deletion of DynamoDB-managed rules with AccessDenied, so only
2150	// AWS support can remove them afterwards.
2151	//
2152	// Observe existence first: `waitForContributorInsightsSettled`
2153	// retries `ResourceNotFoundException` as a control-plane blip (its
2154	// reconcile-side callers run right after the table was created),
2155	// so a table that is already gone would otherwise spin through the
2156	// full ~90s settle budget before this delete can notice.
2157	const tableExists = yield* dynamodb
2158	.describeTable({ TableName: output.tableName })
2159	.pipe(
2160	Effect.as(true),
2161	Effect.catchTag("ResourceNotFoundException", () =>
2162	Effect.succeed(false),
2163	),
2164	);
2165	if (tableExists) {
2166	yield* Effect.gen(function* () {
2167	const insightsStatus = yield* waitForContributorInsightsSettled(
2168	session,
2169	output.tableName,
2170	);
2171	if (insightsStatus !== "DISABLED") {
2172	yield* session.note(
2173	`Table ${output.tableName}: disabling Contributor Insights before delete`,
2174	);
2175	yield* updateTableContributorInsights(output.tableName, false);
2176	yield* waitForContributorInsightsSettled(
2177	session,
2178	output.tableName,
2179	);
2180	}
2181	yield* waitForContributorInsightsRulesDeleted(
2182	session,
2183	output.tableName,
2184	);
2185	}).pipe(
2186	// Table vanished mid-teardown — nothing left to tear down.
2187	Effect.catchTag("ResourceNotFoundException", () => Effect.void),
2188	);
2189	}
2190
2191	let deleteAttempt = 0;
2192
2193	while (true) {
2194	deleteAttempt += 1;
2195	yield* session.note(
2196	`Table ${output.tableName}: deleting (attempt ${deleteAttempt})`,
2197	);
2198
2199	const deleteResult = yield* dynamodb
2200	.deleteTable({
2201	TableName: output.tableName,
2202	})
2203	.pipe(
2204	Effect.timeout(1000),
2205	Effect.as("accepted" as const),
2206	Effect.catchTag("ResourceNotFoundException", () =>
2207	Effect.succeed("already-deleted" as const),
2208	),
2209	Effect.catchTag("ResourceInUseException", () =>
2210	Effect.succeed("delete-gsis-first" as const),
2211	),
2212	Effect.retry({
2213	while: (error) =>
2214	error._tag === "InternalServerError" ||
2215	error._tag === "TimeoutError",
2216	schedule: waitForDeletionConvergence.pipe(
2217	Schedule.tap(({ attempt }) =>
2218	session.note(
2219	`DynamoDB Table provider: deleteTable transient failure for ${output.tableName} on attempt ${deleteAttempt} (${formatPollingElapsed(attempt)})`,
2220	),
2221	),
2222	),
2223	}),
2224	);
2225
2226	if (deleteResult === "accepted") {
2227	yield* session.note(
2228	`DynamoDB Table provider: deleteTable accepted for ${output.tableName}`,
2229	);
2230	break;
2231	}
2232
2233	if (deleteResult === "already-deleted") {
2234	yield* session.note(
2235	`DynamoDB Table provider: table ${output.tableName} already deleted`,
2236	);
2237	return;
2238	}
2239
2240	yield* session.note(
2241	`DynamoDB Table provider: deleteTable blocked for ${output.tableName}; deleting GSIs first`,
2242	);
2243	yield* deleteGlobalSecondaryIndexes(session, output.tableName);
2244	yield* waitForGlobalSecondaryIndexesStable(
2245	session,
2246	output.tableName,
2247	[],
2248	);
2249	yield* waitForTableActive(session, output.tableName);
2250	}
2251
2252	yield* session.note(
2253	`Table ${output.tableName}: waiting for deletion`,
2254	);
2255	yield* waitForTableDeleted(session, output.tableName);
2256	}),
2257	});
2258	}),
2259	);
2260
2261	class TableNotActive extends Data.TaggedError("TableNotActive") {}
2262
2263	class TableIndexesNotStable extends Data.TaggedError("TableIndexesNotStable") {}
2264
2265	class TableStillDeleting extends Data.TaggedError("TableStillDeleting") {}
2266
2267	class MissingStreamViewType extends Data.TaggedError("MissingStreamViewType") {}
2268
2269	class KinesisDestinationNotSettled extends Data.TaggedError(
2270	"KinesisDestinationNotSettled",
2271	) {}
2272
2273	class ContributorInsightsNotSettled extends Data.TaggedError(
2274	"ContributorInsightsNotSettled",
2275	) {}
2276
2277	class KinesisStreamingDestinationFailed extends Data.TaggedError(
2278	"KinesisStreamingDestinationFailed",
2279	)<{
2280	tableName: string;
2281	streamArn: string;
2282	status: string | undefined;
2283	description: string | undefined;
2284	}> {}
2285
2286	class ConflictingStreamViewTypes extends Data.TaggedError(
2287	"ConflictingStreamViewTypes",
2288	)<{
2289	requested: readonly (DynamoDB.StreamViewType | undefined)[];
2290	}> {}