All files / packages/topic/src/writer types.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
import type { CompressionCodec } from '../codec.js'
import type { TX } from '../tx.js'
 
// Status of a single message acknowledged by the server.
export type AckStatus = 'written' | 'skipped' | 'writtenInTx'
 
// A single server acknowledgement, flattened from StreamWriteMessage.WriteResponse.acks.
export type WriteAck = {
	seqNo: bigint
	status: AckStatus
	// Present only for 'written' acks — the partition offset the message landed at.
	offset?: bigint
}
 
// Per-message acknowledgment hook (see TopicWriterOptions.onAck).
export type OnAckCallback = (seqNo: bigint, status: AckStatus) => void
 
// Optional per-message metadata accepted by write().
export type WriteExtra = {
	// User-provided sequence number. Providing it once switches the writer to
	// manual mode: every subsequent message must then also provide a seqNo.
	seqNo?: bigint
	createdAt?: Date
	metadataItems?: Record<string, Uint8Array>
}
 
export type TopicWriterOptions = {
	// Path to the topic to write to, e.g. "/Root/my-topic".
	topic: string
	// Producer identity. Together with seqNo it gives the server-side
	// deduplication key (producer + seqNo), which makes reconnect resends safe.
	// A unique id is generated if omitted.
	producer?: string
	// Transaction identity. When set, every write is tagged with the tx. Stream
	// errors still reconnect transparently — resends are safe inside a tx too
	// (server-side dedup by producer + seqNo); only a failed graceful drain at
	// commit time surfaces to the transaction layer (close() rejects).
	tx?: TX
	// Compression codec. Default RAW.
	codec?: CompressionCodec
 
	// Pin writes to a single partition (mutually exclusive with messageGroupId).
	partitionId?: bigint
	// Route writes by message group (mutually exclusive with partitionId).
	messageGroupId?: string
 
	// Hard cap on the un-acknowledged bytes held in memory; write() throws when a
	// message would exceed it. Default 256MiB.
	maxBufferBytes?: bigint
	// Cap the number of un-acknowledged (in-flight) messages. Default 1000.
	maxInflightCount?: number
	// Background flush cadence in ms — bounds how long a small batch waits. Default 1000.
	flushIntervalMs?: number
 
	// How often to refresh the auth token on the stream. Default 60s.
	updateTokenIntervalMs?: number
	// Force-close deadline for graceful close() before pending messages are dropped.
	// JS-specific safety net; other SDKs bound this by the caller's signal only. Default 30s.
	gracefulShutdownTimeoutMs?: number
	// Terminal reconnect window in ms. Unbounded by default (the writer reconnects
	// forever, waiting for the server / topic); pass a finite value to fail terminally
	// if no successful reconnect happens within it.
	recoveryWindowMs?: number
	// Retry on SCHEME_ERROR (e.g. the topic does not exist yet). Off by default: a
	// missing / mistyped topic fails fast. Enable to wait until the topic is created.
	retryOnSchemeError?: boolean
 
	// Called for every acknowledged message. Errors thrown here are logged and
	// ignored — a throwing callback must never break the writer.
	onAck?: OnAckCallback
}