docs(topic): rewrite README; add Russian README; update homepage

  - Overhaul packages/topic/README.md: quick start, top‑level topic(driver), factory APIs, reader/writer options, transactions, custom codecs, performance notes, exports; note Node.js ≥ 20.19
  - Add packages/topic/README.ru.md with cross‑links
  - Update packages/topic/package.json homepage to the package directory
  - Minor cosmetic cleanup in packages/topic/tests/reader.test.ts

Signed-off-by: Vladislav Polyakov <[email protected]>

Author: polRk

packages/topic/README.md

@@ -1,193 +1,220 @@
 # @ydbjs/topic
 
-The `@ydbjs/topic` package provides high-level, type-safe clients for working with YDB topics (message queues) in JavaScript/TypeScript. It enables efficient streaming reads and writes, partition management, offset commits, and supports compression and custom payload encoding/decoding.
+Read this in Russian: [README.ru.md](README.ru.md)
+
+High‑level, type‑safe clients for YDB Topics (publish–subscribe message streams) in JavaScript/TypeScript. Provides efficient streaming reads and writes, partition session management, offset commits, compression, and transaction‑aware readers/writers.
 
 ## Features
 
-- Streaming topic reader and writer with async iteration
-- Partition session management and offset commit
-- Compression and custom payload encoding/decoding
-- TypeScript support with type definitions
-- Integration with `@ydbjs/core` and `@ydbjs/api`
+- Streaming reader and writer with async iteration
+- Partition session lifecycle hooks and offset commit
+- Pluggable compression codecs (RAW, GZIP, ZSTD; custom via maps)
+- Transaction‑aware read/write helpers
+- First‑class TypeScript types
 
 ## Installation
 
 ```sh
-npm install @ydbjs/core@alpha @ydbjs/topic@alpha
+npm install @ydbjs/topic
 ```
 
-## How It Works
+Requires Node.js >= 20.19.
+
+## Getting Started
 
-- **TopicReader**: Reads messages from a YDB topic as async batches, manages partition sessions, and supports offset commits.
-- **TopicWriter**: Writes messages to a YDB topic, supports batching, compression, and custom encoding.
-- **Integration**: Use with a `Driver` from `@ydbjs/core` for connection management and authentication.
+Two ways to use the client:
 
-## Usage
+- Top‑level client via `topic(driver)`
+- Direct factory functions via subpath imports
 
-### Reading from a Topic
+### Using the top‑level client
 
 ```ts
 import { Driver } from '@ydbjs/core'
-import { TopicReader } from '@ydbjs/topic/reader'
-import { Codec } from '@ydbjs/api/topic'
+import { topic } from '@ydbjs/topic'
 
 const driver = new Driver(process.env['YDB_CONNECTION_STRING']!)
 await driver.ready()
 
-await using reader = new TopicReader(driver, {
-  topic: 'test-topic',
-  consumer: 'test-consumer',
-  maxBufferBytes: 64n * 1024n,
-  compression: {
-    decompress(codec, payload) {
-      if (codec === Codec.GZIP) {
-        return import('node:zlib').then((zlib) => zlib.gunzipSync(payload))
-      } else {
-        throw new Error(`Unsupported codec: ${codec}`)
-      }
-    },
-  },
-})
+const t = topic(driver)
 
-for await (let batch of reader.read({ limit: 50, waitMs: 1000 })) {
-  console.log('received batch', batch.length)
+// Reader
+await using reader = t.createReader({ topic: '/Root/my-topic', consumer: 'my-consumer' })
+for await (const batch of reader.read()) {
+  // Process messages
+  for (const msg of batch) console.log(new TextDecoder().decode(msg.payload))
+  // Commit processed offsets (see performance note below)
   await reader.commit(batch)
 }
+
+// Writer
+await using writer = t.createWriter({ topic: '/Root/my-topic', producer: 'my-producer' })
+writer.write(new TextEncoder().encode('Hello, YDB!'))
+await writer.flush()
 ```
 
-### Writing to a Topic
+### Using direct factories
 
 ```ts
 import { Driver } from '@ydbjs/core'
-import { TopicWriter } from '@ydbjs/topic/writer'
-import { Codec } from '@ydbjs/api/topic'
-import * as zlib from 'node:zlib'
+import { createTopicReader, createTopicTxReader } from '@ydbjs/topic/reader'
+import { createTopicWriter, createTopicTxWriter } from '@ydbjs/topic/writer'
 
 const driver = new Driver(process.env['YDB_CONNECTION_STRING']!)
 await driver.ready()
 
-await using writer = new TopicWriter(driver, {
-  topic: 'test-topic',
-  producer: 'test-producer',
-  maxBufferBytes: 64n * 1024n,
-  flushIntervalMs: 5000,
-  compression: {
-    codec: Codec.GZIP,
-    compress(payload) {
-      return zlib.gzipSync(payload)
-    },
-  },
-})
-
-writer.write(new Uint8Array([1, 2, 3, 4]))
+await using reader = createTopicReader(driver, { topic: '/Root/my-topic', consumer: 'my-consumer' })
+await using writer = createTopicWriter(driver, { topic: '/Root/my-topic', producer: 'my-producer' })
 ```
 
-## Configuration & Options
+## Reader
 
-### TopicReaderOptions
+### Options
 
-Options for configuring a `TopicReader` instance:
+- `topic`: `string | TopicReaderSource | TopicReaderSource[]` — topic path or detailed sources
+- `consumer`: `string` — consumer name
+- `codecMap?`: `Map<Codec | number, CompressionCodec>` — custom codecs for decompression
+- `maxBufferBytes?`: `bigint` — internal buffer cap (default ~4 MiB)
+- `updateTokenIntervalMs?`: `number` — auth token refresh interval (default 60000)
+- `onPartitionSessionStart?`: hook to adjust read/commit offsets per session
+- `onPartitionSessionStop?`: hook on session stop (cleanup/commit)
+- `onCommittedOffset?`: observe commit acknowledgments from server
 
-| Option                    | Type                                                    | Description & Best Practice                                                                                                                         |
-| ------------------------- | ------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `topic`                   | `string \| TopicReaderSource \| TopicReaderSource[]`    | Topic path or array of topic sources. Use a string for a single topic, or an array for multi-topic reading.                                         |
-| `consumer`                | `string`                                                | Consumer name. Use a unique name per logical consumer group.                                                                                        |
-| `maxBufferBytes`          | `bigint`                                                | Max internal buffer size in bytes. Increase for high-throughput, decrease to limit memory usage.                                                    |
-| `updateTokenIntervalMs`   | `number`                                                | How often to update the token (ms). Default: 60000. Lower for short-lived tokens.                                                                   |
-| `compression.decompress`  | `(codec, payload) => Uint8Array \| Promise<Uint8Array>` | Custom decompression function. Use for custom codecs or to enable GZIP/LZ4, etc.                                                                    |
-| `decode`                  | `(payload: Uint8Array) => Payload`                      | Custom payload decoder. Use for JSON, protobuf, or other formats.                                                                                   |
-| `onPartitionSessionStart` | `function`                                              | Called when a partition session starts. Use to set custom read/commit offsets.                                                                      |
-| `onPartitionSessionStop`  | `function`                                              | Called when a partition session stops. Use to commit offsets or cleanup.                                                                            |
-| `onCommittedOffset`       | `function`                                              | Called after offsets are committed. Use for monitoring or logging. For high-throughput, prefer this hook over awaiting `commit()` (see note below). |
+TopicReaderSource supports partition filters and time‑based selectors:
 
-> **Performance Note:**
->
-> The `commit` method can be called without `await` to send commit requests to the server. If you use `await reader.commit(batch)`, your code will wait for the server's acknowledgment before continuing, which can significantly reduce throughput. For best performance in high-load scenarios, avoid awaiting `commit` directly in your main loop. Instead, use the `onCommittedOffset` hook to be notified when the server confirms the commit. This allows your application to process messages at maximum speed while still tracking commit confirmations asynchronously.
+```ts
+const source = {
+  path: '/Root/my-topic',
+  partitionIds: [0n, 1n],
+  // Skip messages older than 5 minutes
+  maxLag: '5m', // number (ms), ms‑string, or Duration
+  // Start from a timestamp
+  readFrom: new Date(Date.now() - 60_000),
+}
+```
 
-#### Example: Custom Decoder and Partition Hooks
+### Reading and committing
 
 ```ts
-await using reader = new TopicReader(driver, {
-  topic: 'test-topic',
-  consumer: 'my-consumer',
-  decode(payload) {
-    return JSON.parse(Buffer.from(payload).toString('utf8'))
-  },
-  onPartitionSessionStart(session, committedOffset, partitionOffsets) {
-    console.log('Partition started', session.partitionId)
-  },
-  onPartitionSessionStop(session, committedOffset) {
-    console.log('Partition stopped', session.partitionId)
-  },
-})
+const t = topic(driver)
+await using reader = t.createReader({ topic: source, consumer: 'svc-a' })
+
+for await (const batch of reader.read({ limit: 100, waitMs: 1000 })) {
+  if (!batch.length) continue // periodic empty batches when no data
+
+  // Handle messages
+  for (const m of batch) doSomething(m)
+
+  // Option A (simple): await commit for each batch
+  await reader.commit(batch)
+
+  // Option B (fast path): fire‑and‑forget commit
+  // void reader.commit(batch)
+}
 ```
 
-### TopicWriterOptions
+Performance note: awaiting `commit()` in the hot path reduces throughput. For high load, prefer fire‑and‑forget plus `onCommittedOffset` to observe confirmations asynchronously.
 
-Options for configuring a `TopicWriter` instance:
+## Writer
 
-| Option                   | Type                                                         | Description & Best Practice                                                                                                                    |
-| ------------------------ | ------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------- |
-| `topic`                  | `string`                                                     | Topic path to write to. Required.                                                                                                              |
-| `producer`               | `string`                                                     | Producer name. Set for idempotency and tracking.                                                                                               |
-| `getLastSeqNo`           | `boolean`                                                    | Get last sequence number before writing. Use for exactly-once or deduplication.                                                                |
-| `allowDuplicates`        | `boolean`                                                    | Allow duplicate messages. Set to true for at-most-once delivery.                                                                               |
-| `updateTokenIntervalMs`  | `number`                                                     | How often to update the token (ms). Default: 60000.                                                                                            |
-| `maxBufferBytes`         | `bigint`                                                     | Max buffer size in bytes. Increase for batching, decrease for low-latency.                                                                     |
-| `maxInflightCount`       | `bigint`                                                     | Max in-flight messages. Tune for throughput vs. memory.                                                                                        |
-| `flushIntervalMs`        | `number`                                                     | Auto-flush interval (ms). Lower for low-latency, higher for throughput.                                                                        |
-| `compression.codec`      | `Codec`                                                      | Compression codec (e.g., GZIP). Use to reduce network usage.                                                                                   |
-| `compression.compress`   | `(payload: Uint8Array) => Uint8Array \| Promise<Uint8Array>` | Custom compression function. Use for custom codecs or advanced compression.                                                                    |
-| `compression.minRawSize` | `bigint`                                                     | Minimum payload size to compress. Avoids compressing small messages.                                                                           |
-| `encode`                 | `(payload: Payload) => Uint8Array`                           | Custom encoder. Use for JSON, protobuf, or other formats.                                                                                      |
-| `onAck`                  | `(seqNo: bigint, status?: string) => void`                   | Called on message acknowledgment. Use for tracking or logging. For high-throughput, prefer this hook over awaiting `write()` (see note below). |
+### Options
 
-> **Performance Note:**
->
-> The `write` method adds messages to an internal buffer and returns a promise that resolves when the server acknowledges the write. If you use `await writer.write(...)`, your code will wait for the server's acknowledgment before continuing, which can significantly reduce throughput. For best performance in high-load scenarios, avoid awaiting `write` directly in your main loop. Instead, use the `onAck` hook to be notified when the server confirms the write. You can tune throughput and latency using `maxBufferBytes`, `maxInflightCount`, and `flushIntervalMs` options to control how quickly messages are sent to the server.
+- `topic`: `string`
+- `tx?`: `TX` — transaction to write within
+- `producer?`: `string` — producer id (auto‑generated if omitted)
+- `codec?`: `CompressionCodec` — compression (default RAW; built‑ins: RAW, GZIP, ZSTD)
+- `maxBufferBytes?`: `bigint` — writer buffer cap (default 256 MiB)
+- `maxInflightCount?`: `number` — max messages in‑flight (default 1000)
+- `flushIntervalMs?`: `number` — periodic flush tick (default 10ms)
+- `updateTokenIntervalMs?`: `number` — auth token refresh interval (default 60000)
+- `retryConfig?(signal)`: tune connection retry strategy
+- `onAck?(seqNo, status)`: callback on message acknowledgment
 
-#### Example: Custom Encoder and Compression
+### Writing messages
 
 ```ts
-await using writer = new TopicWriter(driver, {
-  topic: 'test-topic',
+import { Codec } from '@ydbjs/api/topic'
+
+const t = topic(driver)
+await using writer = t.createWriter({
+  topic: '/Root/my-topic',
   producer: 'json-producer',
-  encode(payload) {
-    return Buffer.from(JSON.stringify(payload), 'utf8')
-  },
-  compression: {
-    codec: Codec.GZIP,
-    compress(payload) {
-      return zlib.gzipSync(payload)
-    },
-  },
+  // Use RAW by default, or provide your own codec implementation.
+  // See "Custom codecs" below for an example.
   onAck(seqNo, status) {
-    console.log('Ack for', seqNo, 'status:', status)
+    console.log('ack', seqNo, status)
   },
 })
 
-writer.write({ foo: 'bar', ts: Date.now() })
+const payload = new TextEncoder().encode(JSON.stringify({ foo: 'bar', ts: Date.now() }))
+const seqNo = writer.write(payload)
+await writer.flush()
+```
+
+`write()` accepts `Uint8Array` only. Encode your own objects/strings as needed.
+
+## Transactions
+
+Run topic reads/writes inside an @ydbjs/query transaction handler and pass the `tx` object that it provides.
+
+- Reader: `createTopicTxReader(driver, { topic, consumer, tx })` or `t.createTxReader({ ..., tx })`. The reader tracks read offsets and automatically issues updateOffsetsInTransaction on commit.
+- Writer: `createTopicTxWriter(tx, driver, { topic, ... })` or `t.createTxWriter(tx, { ... })`. The writer ensures a flush before the transaction commits.
+
+```ts
+import { query } from '@ydbjs/query'
+import { createTopicTxReader } from '@ydbjs/topic/reader'
+import { createTopicTxWriter } from '@ydbjs/topic/writer'
+
+const qc = query(driver)
+
+await qc.transaction(async (tx, signal) => {
+  // Tx‑aware reader: offsets are bound to the transaction commit
+  await using reader = createTopicTxReader(driver, { topic: '/Root/my-topic', consumer: 'svc-a', tx })
+  for await (const batch of reader.read({ signal })) {
+    // process batch...
+  }
+
+  // Tx‑aware writer: flushes before commit
+  await using writer = createTopicTxWriter(tx, driver, { topic: '/Root/my-topic', producer: 'p1' })
+  writer.write(new TextEncoder().encode('message'))
+})
 ```
 
-## API
+Note: `tx` comes from the Query layer and exposes the required hooks; Topic clients integrate with them automatically.
 
-### TopicReader
+## Custom codecs
 
-- Reads messages from a topic as async batches
-- Supports custom decompression and decoding
-- Partition session hooks: `onPartitionSessionStart`, `onPartitionSessionStop`, `onCommittedOffset`
-- Offset commit with `commit()`
+Reader supports custom decompression through `codecMap` and writer via a `CompressionCodec` instance.
 
-### TopicWriter
+```ts
+import { Codec } from '@ydbjs/api/topic'
+import * as zlib from 'node:zlib'
 
-- Writes messages to a topic, supports batching and compression
-- Custom encoding and compression
-- Ack callback: `onAck`
+const MyGzip = {
+  codec: Codec.GZIP,
+  compress: (p: Uint8Array) => zlib.gzipSync(p),
+  decompress: (p: Uint8Array) => zlib.gunzipSync(p),
+}
+
+await using reader = createTopicReader(driver, {
+  topic: '/Root/custom',
+  consumer: 'c1',
+  codecMap: new Map([[Codec.GZIP, MyGzip]]),
+})
+
+await using writer = createTopicWriter(driver, {
+  topic: '/Root/custom',
+  producer: 'p1',
+  codec: MyGzip,
+})
+```
 
-### Types
+## Exports
 
-- `TopicMessage<Payload>`: Message structure for topic payloads
-- `TopicReaderOptions`, `TopicWriterOptions`: Configuration options for reader and writer
+- `@ydbjs/topic`: `topic(driver)` and types
+- `@ydbjs/topic/reader`: `createTopicReader`, `createTopicTxReader`, reader types
+- `@ydbjs/topic/writer`: `createTopicWriter`, `createTopicTxWriter`, writer types
+- `@ydbjs/topic/writer2`: experimental state‑machine writer (API subject to change)
 
 ## License