TopicService

TopicService manages the topic catalog: the registered event topics, their payload schemas and the agents subscribed to them. It is a direct import, not a client namespace, and its functions return plain values rather than LTApiResult.

import { TopicService } from '@hotmeshio/long-tail';

See Agents and events.

listTopics

const result = await TopicService.listTopics({
  category: 'task',
  search: 'failed',
  limit: 50,
  offset: 0,
});
// result.topics: TopicCatalogEntry[]
// result.total: number

List topics with optional filters.

Parameters

Field	Type	Required	Description
category	string	No	Filter by category
search	string	No	Search name or description
limit	number	No	Max results (default 50)
offset	number	No	Pagination offset

Returns: { topics: TopicCatalogEntry[], total: number }

getTopic

const topic = await TopicService.getTopic('system.task.*.created');
// topic.payload_schema: JSON Schema for event.data
// topic.subscribers: agents whose patterns match this topic

Get one topic with its subscriber list. Subscribers match by NATS-style patterns, so an agent subscribed to system.task.> appears under every task topic.

Parameters

Field	Type	Required	Description
topic	string	Yes	Topic name (the first argument)

Returns: TopicCatalogEntry & { subscribers: TopicSubscriber[] }, or null.

createTopic

const entry = await TopicService.createTopic({
  topic: 'app.vendor.orders.created',
  description: 'Fired when a new order is placed.',
  category: 'app',
  payload_schema: {
    type: 'object',
    properties: {
      orderId: { type: 'string' },
      total: { type: 'number' },
    },
  },
  tags: ['orders', 'lifecycle'],
});

Register a topic in the catalog.

Parameters

Field	Type	Required	Description
topic	string	Yes	Unique topic name
category	string	Yes	Topic category
description	string	No	Description
payload_schema	object	No	JSON Schema for event.data
example_payload	object	No	Example event.data
source	string	No	Source identifier (default 'app')
tags	string[]	No	Tags for filtering

Returns: TopicCatalogEntry. Throws a Postgres unique violation when the topic exists.

updateTopic

const updated = await TopicService.updateTopic('app.vendor.orders.created', {
  description: 'Updated description',
  tags: ['orders', 'critical'],
});

Partially update a topic’s metadata.

Parameters

Field	Type	Required	Description
topic	string	Yes	Topic name (the first argument)
data	Partial<TopicCatalogEntry>	Yes	Fields to change (the second argument)

Returns: TopicCatalogEntry | null

deleteTopic

const deleted = await TopicService.deleteTopic('app.vendor.orders.created');

Delete a topic. System topics (source = 'system') are kept.

Parameters

Field	Type	Required	Description
topic	string	Yes	Topic name (the first argument)

Returns: boolean

seedTopic

const inserted = await TopicService.seedTopic({
  topic: 'app.orders.created',
  description: 'New order placed',
  category: 'app',
  tags: ['orders'],
});
// true if new, false if it already existed

Insert a topic if absent. Used for first-boot seeding.

Parameters: the fields of createTopic; description is required here.

Returns: boolean

resetTopic

await TopicService.resetTopic({
  topic: 'app.orders.created',
  description: 'New order placed (updated)',
  category: 'app',
  payload_schema: { type: 'object', properties: { orderId: { type: 'string' } } },
});

Upsert from config. Overwrites description, category, schema and tags on every call. Used when reset: true is set in static config. See Code-owned config.

Parameters: the same fields as createTopic.

Returns: void

upsertTopicOnPublish

await TopicService.upsertTopicOnPublish(
  'app.vendor.orders.error',
  { orderId: '123', error: 'timeout' },  // example data
  'order-processor',                      // source
);

Register or update a topic when an event is published. The first publish creates the entry; later publishes update last_seen_at and example_payload. The publish_event MCP tool calls it for you; call it yourself only on manual publish paths.

Parameters

Field	Type	Required	Description
topic	string	Yes	Topic name (first argument)
data	object	No	Example payload (second argument)
source	string	No	Source identifier (third argument)

Returns: void

Types

interface TopicCatalogEntry {
  topic: string;
  description?: string;
  category: string;
  payload_schema?: Record<string, any>;
  example_payload?: Record<string, any>;
  source: string;            // 'system', 'config', 'app', 'mcp-tool'
  tags: string[];
  managed: boolean;          // true when static config (reset: true) owns the topic
  subscriber_count?: number; // populated by listTopics
  last_seen_at?: string;     // ISO 8601, updated on each publish
  created_at: string;
  updated_at: string;
}

interface TopicSubscriber {
  id: string;
  agent_id: string;
  agent_name: string;
  topic: string;             // the subscription pattern
  reaction_type: string;     // 'durable', 'pipeline', 'mcp_query'
}

Seeding helpers

Two seeding functions are exported from the package root for custom startup flows. start() calls them for you.

import { seedSystemTopics, seedConfigTopics } from '@hotmeshio/long-tail';

// Seed the built-in system topics
await seedSystemTopics();

// Seed declared topics (reset: true overwrites from config)
await seedConfigTopics([
  { topic: 'app.orders.created', description: 'Order placed', reset: true },
]);