hub

Documentation

Overview

Package hub holds the durable source of truth for cross-channel conversations. It lets messages from any provider (Telegram, Slack, WhatsApp) and the local CLI flow into a single shared conversation per principal, so a thread started on one channel continues seamlessly on another until the user starts a new session.

Index

• Variables
• func DefaultDBPath() (string, error)
• type Binding
• type Broker
• type Manager
  • func NewManager(store Store, logger *zap.Logger, bufSize int) *Manager
  • func OpenDefault(ctx context.Context, logger *zap.Logger) (*Manager, error)
  • func (m *Manager) Append(ctx context.Context, ev models.ConversationEvent) (models.ConversationEvent, error)
  • func (m *Manager) Subscribe(ctx context.Context, convID string, sinceSeq int64) (<-chan models.ConversationEvent, error)
• type SQLiteStore
  • func OpenSQLiteStore(ctx context.Context, path string, logger *zap.Logger) (*SQLiteStore, error)
  • func (s *SQLiteStore) AllSettings(ctx context.Context) (map[string]string, error)
  • func (s *SQLiteStore) Append(ctx context.Context, ev models.ConversationEvent) (models.ConversationEvent, error)
  • func (s *SQLiteStore) Bind(ctx context.Context, platform, userID, principal string) error
  • func (s *SQLiteStore) Close() error
  • func (s *SQLiteStore) DeleteSetting(ctx context.Context, key string) error
  • func (s *SQLiteStore) GetSetting(ctx context.Context, key string) (string, bool, error)
  • func (s *SQLiteStore) ListBindings(ctx context.Context, principal string) ([]Binding, error)
  • func (s *SQLiteStore) NewConversation(ctx context.Context, principal string) (string, error)
  • func (s *SQLiteStore) OwnerOf(ctx context.Context, convID string) (string, error)
  • func (s *SQLiteStore) PurgeIdle(ctx context.Context, olderThan time.Duration) (int, error)
  • func (s *SQLiteStore) Read(ctx context.Context, convID string, sinceSeq int64, limit int) ([]models.ConversationEvent, error)
  • func (s *SQLiteStore) Resolve(ctx context.Context, principal string) (string, error)
  • func (s *SQLiteStore) ResolvePrincipal(ctx context.Context, platform, userID string) (string, error)
  • func (s *SQLiteStore) SetSetting(ctx context.Context, key, value string) error
• type Store

Variables

var ErrUnboundChannel = errors.New("hub: channel identity is not bound to a principal")

ErrUnboundChannel is returned when a channel identity has no principal binding. The Hub quarantines such identities rather than guessing an owner, preventing one person's messages from leaking into another's conversation.

Functions

func DefaultDBPath

func DefaultDBPath() (string, error)

DefaultDBPath returns the conversation hub database path: CHATCLI_HUB_DB when set, otherwise ~/.chatcli/hub.db. The parent directory is created.

Types

type Binding

type Binding struct {
	Platform  string
	UserID    string
	Principal string
}

Binding maps a per-platform channel identity to a principal.

type Broker

type Broker interface {
	Store
	// Subscribe returns a stream of a conversation's events with Seq > sinceSeq:
	// first the persisted backlog, then live events as they are appended. The
	// stream closes when ctx is canceled or when the consumer falls too far
	// behind (see Manager overflow handling), at which point the caller should
	// resubscribe with the last Seq it saw to resync.
	Subscribe(ctx context.Context, convID string, sinceSeq int64) (<-chan models.ConversationEvent, error)
}

Broker is a Store that also supports live tailing of a conversation.

type Manager

type Manager struct {
	Store
	// contains filtered or unexported fields
}

Manager wraps a Store with an in-memory fan-out layer so multiple frontends can live-tail the same conversation. Appends are persisted first, then published to subscribers; durability never depends on the fan-out.

func NewManager

func NewManager(store Store, logger *zap.Logger, bufSize int) *Manager

NewManager wraps store. bufSize bounds each subscriber's live buffer; a consumer that lets it fill is dropped and must resync (back-pressure that protects the Hub from a stalled client). bufSize <= 0 defaults to 256.

func OpenDefault

func OpenDefault(ctx context.Context, logger *zap.Logger) (*Manager, error)

OpenDefault opens the hub at DefaultDBPath wrapped in a fan-out Manager so connected frontends can live-tail. The tail buffer is tunable via CHATCLI_HUB_TAIL_BUFFER. Both the gRPC server and the gateway daemon open the same database file, so a conversation is shared across every channel; for real-time cross-process push, co-locate the gateway in the hub server process (one in-memory Manager).

func (*Manager) Append

func (m *Manager) Append(ctx context.Context, ev models.ConversationEvent) (models.ConversationEvent, error)

Append persists the event via the wrapped Store, then publishes it to live subscribers. A persistence failure is returned without publishing.

func (*Manager) Subscribe

func (m *Manager) Subscribe(ctx context.Context, convID string, sinceSeq int64) (<-chan models.ConversationEvent, error)

Subscribe implements Broker.

type SQLiteStore

type SQLiteStore struct {
	// contains filtered or unexported fields
}

SQLiteStore is the WAL-backed implementation of Store. It is safe for concurrent use: reads go through the connection pool, writes are serialized by wmu so the embedded single-writer never returns SQLITE_BUSY to callers.

func OpenSQLiteStore

func OpenSQLiteStore(ctx context.Context, path string, logger *zap.Logger) (*SQLiteStore, error)

OpenSQLiteStore opens (creating if needed) the Hub database at path with WAL journaling and runs migrations. A nil logger is replaced with a no-op.

func (*SQLiteStore) AllSettings

func (s *SQLiteStore) AllSettings(ctx context.Context) (map[string]string, error)

AllSettings returns every stored runtime setting.

func (*SQLiteStore) Append

func (s *SQLiteStore) Append(ctx context.Context, ev models.ConversationEvent) (models.ConversationEvent, error)

Append writes an event, assigning its Seq. With a ClientMsgID it is idempotent: a repeat returns the previously stored event.

func (*SQLiteStore) Bind

func (s *SQLiteStore) Bind(ctx context.Context, platform, userID, principal string) error

Bind upserts a channel-identity → principal binding.

func (*SQLiteStore) Close

func (s *SQLiteStore) Close() error

Close releases the underlying database.

func (*SQLiteStore) DeleteSetting

func (s *SQLiteStore) DeleteSetting(ctx context.Context, key string) error

DeleteSetting removes a runtime setting so resolution falls back to env/default.

func (*SQLiteStore) GetSetting

func (s *SQLiteStore) GetSetting(ctx context.Context, key string) (string, bool, error)

GetSetting returns a runtime setting and whether it was present. Settings live in the shared database, so a value set by the CLI is read live by the gateway (and vice versa) without an env var or a restart.

func (*SQLiteStore) ListBindings

func (s *SQLiteStore) ListBindings(ctx context.Context, principal string) ([]Binding, error)

ListBindings returns bindings, optionally filtered to one principal.

func (*SQLiteStore) NewConversation

func (s *SQLiteStore) NewConversation(ctx context.Context, principal string) (string, error)

NewConversation always rotates the pointer to a fresh conversation.

func (*SQLiteStore) OwnerOf

func (s *SQLiteStore) OwnerOf(ctx context.Context, convID string) (string, error)

OwnerOf returns the principal owning a conversation.

func (*SQLiteStore) PurgeIdle

func (s *SQLiteStore) PurgeIdle(ctx context.Context, olderThan time.Duration) (int, error)

PurgeIdle deletes conversations whose most recent activity is older than olderThan (and their events), reclaiming space from abandoned threads. A conversation with no events falls back to its creation time. Returns the number of conversations removed.

func (*SQLiteStore) Read

func (s *SQLiteStore) Read(ctx context.Context, convID string, sinceSeq int64, limit int) ([]models.ConversationEvent, error)

Read returns events with Seq > sinceSeq, ordered ascending.

func (*SQLiteStore) Resolve

func (s *SQLiteStore) Resolve(ctx context.Context, principal string) (string, error)

Resolve returns the active conversation for a principal, creating one on first contact. The read fast-path needs no lock; creation is safe across processes sharing the database (the local CLI and the gateway daemon), since the write lock only serializes within one process — see resolveCreateLocked.

func (*SQLiteStore) ResolvePrincipal

func (s *SQLiteStore) ResolvePrincipal(ctx context.Context, platform, userID string) (string, error)

ResolvePrincipal maps a channel identity to its principal.

func (*SQLiteStore) SetSetting

func (s *SQLiteStore) SetSetting(ctx context.Context, key, value string) error

SetSetting upserts a runtime setting.

type Store

type Store interface {
	// Resolve returns the active conversation id for a principal, creating a
	// fresh conversation (and pointer) the first time it is seen.
	Resolve(ctx context.Context, principal string) (convID string, err error)

	// NewConversation rotates the active-conversation pointer for a principal
	// to a brand-new conversation and returns its id. This is what /newsession
	// triggers; every channel resolving that principal afterwards lands on the
	// new conversation.
	NewConversation(ctx context.Context, principal string) (convID string, err error)

	// Append writes an event, assigning its Seq, and returns the stored event.
	// If ev.ClientMsgID is non-empty and already present in the conversation,
	// the existing event is returned unchanged (idempotent retry).
	Append(ctx context.Context, ev models.ConversationEvent) (models.ConversationEvent, error)

	// Read returns events of a conversation with Seq strictly greater than
	// sinceSeq, ordered by Seq ascending, capped at limit (limit <= 0 = all).
	Read(ctx context.Context, convID string, sinceSeq int64, limit int) ([]models.ConversationEvent, error)

	// ResolvePrincipal maps a channel identity to its principal, or returns
	// ErrUnboundChannel when the identity has not been bound.
	ResolvePrincipal(ctx context.Context, platform, userID string) (principal string, err error)

	// Bind associates a channel identity with a principal (idempotent upsert).
	Bind(ctx context.Context, platform, userID, principal string) error

	// ListBindings returns bindings, optionally filtered to one principal
	// (empty principal = all).
	ListBindings(ctx context.Context, principal string) ([]Binding, error)

	// OwnerOf returns the principal that owns a conversation, for authorization
	// checks: a subscriber must own the conversation it tails.
	OwnerOf(ctx context.Context, convID string) (principal string, err error)

	// PurgeIdle deletes conversations idle longer than olderThan (and their
	// events), keeping the hub bounded. The active conversation of each
	// principal is never purged. Returns how many were removed.
	PurgeIdle(ctx context.Context, olderThan time.Duration) (int, error)

	// GetSetting / SetSetting / DeleteSetting / AllSettings manage runtime hub
	// settings stored in the shared database, so changes made by one process
	// (the CLI) are seen live by another (the gateway daemon).
	GetSetting(ctx context.Context, key string) (value string, ok bool, err error)
	SetSetting(ctx context.Context, key, value string) error
	DeleteSetting(ctx context.Context, key string) error
	AllSettings(ctx context.Context) (map[string]string, error)

	// Close releases the underlying database.
	Close() error
}

Store is the durable, concurrency-safe source of truth for cross-channel conversations: an append-only event log per conversation, a per-principal "active conversation" pointer, and channel-identity → principal bindings.

Reads are safe for concurrent use. Writes are serialized internally, so a Telegram adapter and a notebook CLI may append to the same conversation without clobbering one another; each append receives a server-assigned, monotonically increasing Seq.