### Complete Worker Example for Outbox Events

Source: https://context7.com/yornaath/service-outbox/llms.txt

This example demonstrates a message relay worker that forwards outbox events to an external message broker like Kafka or RabbitMQ. It requires initialization of the ServiceOutbox, connection to MongoDB, and uses a cursor for resuming from the last checkpoint. Ensure environment variables like MONGODB_URI are set.

```typescript
import mongoose from "mongoose";
import { tap, runEffects } from "@most/core";
import { newDefaultScheduler } from "@most/scheduler";
import { ServiceOutbox, TOutboxMessage } from "@piing/service-outbox";

// Message types
type OrderEvent =
  | { type: "order_created"; data: { orderId: string; amount: number }}
  | { type: "order_cancelled"; data: { orderId: string; reason: string }};

// Initialize
const outbox = new ServiceOutbox<OrderEvent>("OrderEvents", {
  "order_created": {
    data: {
      orderId: { type: String, required: true },
      amount: { type: Number, required: true }
    }
  },
  "order_cancelled": {
    data: {
      orderId: { type: String, required: true },
      reason: { type: String, required: true }
    }
  }
});

// Relay worker
const startRelayWorker = async () => {
  await mongoose.connect(process.env.MONGODB_URI!, {
    useNewUrlParser: true,
    useUnifiedTopology: true
  });

  // Resume from last checkpoint
  const cursor = await getCursorFromRedis() || new Date(0);

  const tail = outbox.tail({ created: { $gt: cursor }});

  const relay$ = tap(async (message: TOutboxMessage<OrderEvent>) => {
    // Forward to Kafka/RabbitMQ/etc
    await kafka.send({
      topic: "order-events",
      messages: [{
        key: message.type,
        value: JSON.stringify(message.data),
        timestamp: message.created.toISOString()
      }]
    });

    // Checkpoint after successful delivery
    await saveCursorToRedis(message.created);
  }, tail.stream$);

  runEffects(relay$, newDefaultScheduler());
  tail.open();

  console.log("Relay worker started, consuming from:", cursor);
  return tail;
};

startRelayWorker().catch(console.error);

```

--------------------------------

### Tail Service Outbox and Publish Events to Kafka

Source: https://github.com/yornaath/service-outbox/blob/master/readme.md

Sets up a worker to tail the service outbox for new events starting from a specified cursor date. Published events are pushed to Kafka, and the latest published event's timestamp is recorded to ensure idempotency on service restarts. Note that a crash between publishing and recording the timestamp could lead to duplicate messages.

```typescript
const worker = async() => {

  const cursor: Date = await getLatestPublishedServiceCreatedDate()

  const [outbox$, scheduler, close] = outbox.tail(cursor)

  const published$ = tap(async (orderServiceMessage) => {
    await publishToKafka("orders", orderServiceMessage.type, orderServiceMessage.data)
      // Note: if the server where to crash here, its a chance the message can be sent twice.
      // But that is a better option than loosing a message
    await setLatestPublishedServiceCreatedDate(orderServiceMessage.created)
  }, outbox$,)

  runEffects(published$, scheduler)

}
```

--------------------------------

### Configuration Options

Source: https://context7.com/yornaath/service-outbox/llms.txt

Configuration settings for the ServiceOutbox instance to control collection size and polling behavior.

```APIDOC
## TServiceOutboxConfiguration

### Description
Customize the outbox collection behavior during initialization.

### Parameters
- **size** (number) - Optional - Capped collection size in bytes (default: 2KB).
- **max** (number) - Optional - Maximum number of documents in the collection (default: 1000).
- **awaitInterval** (number) - Optional - Polling interval in milliseconds when the collection is empty (default: 200ms).
```

--------------------------------

### Initialize ServiceOutbox with Schema Validation

Source: https://context7.com/yornaath/service-outbox/llms.txt

Demonstrates how to create a ServiceOutbox instance with message type definitions and schema validation for MongoDB. Configure collection options like size, max documents, and polling interval.

```typescript
import { ServiceOutbox } from "@piing/service-outbox";
import mongoose from "mongoose";

// Define message types with strong typing
type OrderCreatedMessage = {
  type: "order_created";
  data: {
    orderId: string;
    customerId: string;
    amount: number;
  };
};

type OrderShippedMessage = {
  type: "order_shipped";
  data: {
    orderId: string;
    trackingNumber: string;
  };
};

type Message = OrderCreatedMessage | OrderShippedMessage;

// Create outbox with schema validation
const outbox = new ServiceOutbox<Message>("OrderServiceOutbox", {
  "order_created": {
    data: {
      orderId: { type: String, required: true },
      customerId: { type: String, required: true },
      amount: { type: Number, required: true }
    }
  },
  "order_shipped": {
    data: {
      orderId: { type: String, required: true },
      trackingNumber: { type: String, required: true }
    }
  }
}, {
  size: 1024 * 1024,  // 1MB capped collection size
  max: 10000,         // Maximum 10000 documents
  awaitInterval: 200  // Poll interval in ms when collection is empty
});
```

--------------------------------

### Subscribe to Outbox Events with tail()

Source: https://context7.com/yornaath/service-outbox/llms.txt

Demonstrates consuming messages from the outbox using tailable cursors, including filtering by timestamp and running concurrent consumers.

```typescript
import { tap, runEffects } from "@most/core";
import { newDefaultScheduler } from "@most/scheduler";
import { ServiceOutbox } from "@piing/service-outbox";

// Basic tail - consume all messages from the outbox
const consumeAllMessages = async () => {
  const tail = outbox.tail();

  const processed$ = tap((message) => {
    console.log(`Received: ${message.type}`, message.data);
    console.log(`Created at: ${message.created}`);  // Each message has a 'created' timestamp
  }, tail.stream$);

  runEffects(processed$, newDefaultScheduler());

  // Start consuming messages
  tail.open();

  // Later, gracefully close the tail
  // await tail.close();
};

// Tail with query filter - resume from a specific timestamp
const resumeFromCursor = async () => {
  // Get the last processed message timestamp from your persistence layer
  const lastProcessedDate = await getLastProcessedTimestamp();

  const tail = outbox.tail({
    created: { $gt: lastProcessedDate }
  });

  const processed$ = tap(async (message) => {
    // Process the message (e.g., publish to Kafka)
    await publishToKafka("orders", message.type, message.data);

    // Update cursor after successful processing
    // Note: if crash occurs here, message may be sent twice (at-least-once delivery)
    await saveLastProcessedTimestamp(message.created);
  }, tail.stream$);

  runEffects(processed$, newDefaultScheduler());
  tail.open();

  return tail;  // Return for later cleanup
};

// Concurrent tails - multiple consumers on same outbox
const multipleTails = async () => {
  const tailA = outbox.tail({ created: { $gt: new Date() }});
  const tailB = outbox.tail({ created: { $gt: new Date() }});

  // Both tails receive the same messages independently
  runEffects(tap(m => console.log("Consumer A:", m.type), tailA.stream$), newDefaultScheduler());
  runEffects(tap(m => console.log("Consumer B:", m.type), tailB.stream$), newDefaultScheduler());

  tailA.open();
  tailB.open();
};
```

--------------------------------

### tail() - Subscribe to Outbox Events

Source: https://context7.com/yornaath/service-outbox/llms.txt

Creates a tailable cursor stream to consume messages from the outbox collection. Supports filtering via query parameters and concurrent consumption.

```APIDOC
## tail(filter?)

### Description
Creates a tailable cursor stream to consume messages from the outbox collection using @most/core reactive streams.

### Parameters
#### Query Parameters
- **filter** (Object) - Optional - A MongoDB-style query object to filter messages (e.g., { created: { $gt: date } }).

### Request Example
```typescript
const tail = outbox.tail({
  created: { $gt: new Date('2023-01-01') }
});
```

### Response
- **TOutboxTail** (Interface) - Returns an object containing stream control methods and the reactive stream.
```

--------------------------------

### Configure ServiceOutbox

Source: https://context7.com/yornaath/service-outbox/llms.txt

Configures the capped collection settings and schema definitions for the ServiceOutbox instance.

```typescript
import { ServiceOutbox, TServiceOutboxConfiguration } from "@piing/service-outbox";

const config: TServiceOutboxConfiguration = {
  size: 1024 * 1024 * 10,  // 10MB capped collection (default: 2KB)
  max: 50000,               // Max 50000 documents (default: 1000)
  awaitInterval: 500        // Poll every 500ms when empty (default: 200ms)
};

const outbox = new ServiceOutbox<Message>("HighVolumeOutbox", {
  "event_a": { data: { value: { type: String, required: true }}},
  "event_b": { data: { count: { type: Number, required: true }}}
}, config);
```

--------------------------------

### TOutboxTail Interface and Lifecycle Management

Source: https://context7.com/yornaath/service-outbox/llms.txt

Defines the structure of the tail object and shows how to implement graceful shutdown handling for message workers.

```typescript
import { Stream } from "@most/types";

interface TOutboxTail<T, TM> {
  open(): void;      // Start consuming messages
  close(): void;     // Stop consuming and close cursor
  stream$: Stream<TM>;  // @most/core Stream of messages
}

// Usage pattern with proper lifecycle management
const messageWorker = async () => {
  const tail = outbox.tail();

  // Setup stream processing before opening
  const errorHandled$ = tap((message) => {
    try {
      processMessage(message);
    } catch (error) {
      console.error("Processing failed:", error);
    }
  }, tail.stream$);

  runEffects(errorHandled$, newDefaultScheduler());

  // Open the tail to start receiving messages
  tail.open();

  // Graceful shutdown handler
  process.on("SIGTERM", async () => {
    console.log("Shutting down message worker...");
    await tail.close();
    process.exit(0);
  });
};
```

--------------------------------

### Batch Publish Multiple Messages with Auto-Commit

Source: https://context7.com/yornaath/service-outbox/llms.txt

Illustrates publishing multiple messages to the outbox in a single atomic operation using `outbox.put()` with an array of messages. The `autoCommit: true` option commits the transaction automatically upon successful insertion.

```typescript
// Batch publish multiple messages
const publishBatchMessages = async (messages: Message[]) => {
  const session = await mongoose.startSession();
  session.startTransaction();

  // Put multiple messages in a single transaction
  const result = await outbox.put([
    { type: "order_created", data: { orderId: "123", customerId: "c1", amount: 100 }},
    { type: "order_shipped", data: { orderId: "122", trackingNumber: "TRACK123" }}
  ], session, { autoCommit: true });  // Auto-commit the transaction

  return result;  // Returns array of created documents with 'created' timestamp
};
```

--------------------------------

### TOutboxTail Interface

Source: https://context7.com/yornaath/service-outbox/llms.txt

The interface returned by the tail() method, providing control over the stream lifecycle.

```APIDOC
## TOutboxTail Interface

### Description
Provides methods to manage the lifecycle of the outbox stream.

### Methods
- **open()**: void - Starts consuming messages from the outbox.
- **close()**: void - Stops consuming and closes the cursor.

### Properties
- **stream$**: Stream<TM> - The @most/core reactive stream of messages.
```

--------------------------------

### Atomically Publish Message within Transaction

Source: https://context7.com/yornaath/service-outbox/llms.txt

Shows how to use the `put()` method to atomically store a message in the outbox collection as part of a MongoDB transaction. This ensures that the order creation and event publishing are an all-or-nothing operation.

```typescript
import mongoose from "mongoose";
import { ServiceOutbox } from "@piing/service-outbox";
import Order from "./models/Order";

// Atomic operation: create order and publish event in same transaction
const createOrder = async (customerId: string, amount: number) => {
  const session = await mongoose.startSession();
  session.startTransaction();

  try {
    // Create the order in your orders collection
    const order = await Order.create([{ 
      customerId,
      amount,
      status: "pending"
    }], { session });

    // Publish event to outbox - both operations are atomic
    await outbox.put({
      type: "order_created",
      data: {
        orderId: order[0]._id.toString(),
        customerId,
        amount
      }
    }, session);

    await session.commitTransaction();
    return order[0];
  } catch (error) {
    await session.abortTransaction();
    throw error;
  } finally {
    session.endSession();
  }
};
```

--------------------------------

### Define Service Outbox with TypeScript and Mongoose Schema Validation

Source: https://github.com/yornaath/service-outbox/blob/master/readme.md

Defines a strongly-typed Service Outbox instance using TypeScript generics and integrates with Mongoose for schema validation. Ensure the message types and their corresponding Mongoose schemas are kept in sync.

```typescript
import { ServiceOutbox } from "@piing/service-outbox"
import Order from "./models/Order"


// Defining the types of messages this service can publish.

type OrderCreatedMessage = {
  type: "order_created"
  data: {
    ordernr: number
  }
}

type OrderCompletedMessage = {
  type: "order_completed"
  data: {
    ordernr: number
  }
}

type Message = OrderCreatedMessage | OrderCompletedMessage

// Defining the outbox has strong typing and mongoose schema validation.
// Key is the type of message and the object associated is a mongoose.Schema
// Challenge is keeping the types and Schema in sync

const outbox = new ServiceOutbox<Message>("OrderServiceOutbox",{
  "order_created": {
    data: {
      ordernr: {
        type: Number,
        required: true
      }
    }
  },
  "order_completed": {
    data: {
      ordernr: {
        type: Number,
        required: true
      }
    }
  }
})
```

--------------------------------

### Atomic Order Creation and Outbox Message Publication

Source: https://github.com/yornaath/service-outbox/blob/master/readme.md

Performs an atomic operation to create an order and publish an 'order_created' event using a MongoDB transaction. This ensures that both the order creation and the event publication succeed or fail together.

```typescript
const makeOrder = async() => {
  const session = await mongoose.startSession()

  session.startTransaction()

  const order = await Order.create(..., { session })

  await outbox.put({
    type: "order_created",
    data: {
      ordernr: order.nr
    }
  }), session)

  await session.commitTransaction()

  return order
}
```

--------------------------------

### ServiceOutbox.put()

Source: https://context7.com/yornaath/service-outbox/llms.txt

Publishes one or more messages to the outbox collection atomically within a MongoDB transaction session.

```APIDOC
## put(messages, session, options)

### Description
Atomically stores one or more messages in the outbox collection within a MongoDB transaction session to ensure data consistency.

### Parameters
#### Request Body
- **messages** (Message | Message[]) - Required - A single message object or an array of message objects to be published.
- **session** (ClientSession) - Required - The MongoDB transaction session to associate with the operation.
- **options** (Object) - Optional - Configuration options such as { autoCommit: boolean }.

### Request Example
{
  "type": "order_created",
  "data": {
    "orderId": "123",
    "customerId": "c1",
    "amount": 100
  }
}

### Response
#### Success Response (200)
- **result** (Array) - Returns an array of created documents including a 'created' timestamp.
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.