Collector

A gRPC + Protocol Buffers framework for building distributed, dynamic RPC systems with built-in service discovery, type safety, and a powerful ORM for protobuf messages.

HUMAN NOTE

Contact fred if you want to try using this! Smash that star button to get updates as we try to get it production ready, ~January 2026.

What is Collector?

Collector is a distributed programming platform that combines: - Service Registry: Type-safe registration and validation of gRPC services - Collections: ORM-like storage for protobuf messages with full-text search - Dynamic Dispatch: Transparent distributed RPC routing across clusters - Reflection & Discovery: Runtime service introspection and dynamic invocation

It enables you to register and update protobuf messages and gRPC services at runtime, create "Collections" (tables + API servers) of any proto type, and dynamically dispatch RPC calls across a distributed system—all with strong typing and validation.

Tell me more!

Collections are generic protobuf container for structured data, backed by sqlite. A collection is an ordered list of records of a single proto message type, filterable labels, and create/update metadata.

Namespaces are the core of collector's multitenancy model. A collection belongs to a single namespace, but the services respect a hierarchical namespacing model that keeps data/types/services/everything else separate across namespaces.

CollectionService is implements generic CRUD and Search APIs for Collections, plus the ability to invoke custom rpcs on the Collections. These are all provided automatically.

CollectionRepo is a Collection of Collections, "controller" of collection service and other internal/registered grpc servers and collections for this "collector". Also handles backups, clones/fetches, etc.

CollectiveDispatcher implements Connect, Dispatch, and Serve rpcs across Collectors. This is what turns Collector into a node in a mesh. It also allows execution to move to data or find available compute dynamically.

CollectorRegistry provides a registry of proto messages and grpcs with some reflection functionality. The rpc/message registry are what make this work as an "agent mesh" - it allows agents to interact with remote collectors, or remote collectors to interact with the local agent, in more compact/api-based interfaces than just text.Z

Using it

CollectionService alone is kind of like a protobuf ORM/CRUD server with search. CollectionRepo also gets you backups and db operations, including backups.

Paired with statue these sqlite-based "collections" can be used as a portable/snapshotted generic container for structured data, that you can distribute and query on a static site!

Enable CollectiveDispatcher and CollectorRegistry to get something with all the right characteristics of a node in an "agent mesh" - hierarchical multitenancy, data/service/type discovery, dynamic tool creation, interoperability with humans or traditional stateful instances/stateless services (which can also run collector!). Note: you should probably run this in a secure sandbox.

TL;DR: Collector is meant to serve as a node in an agent mesh, providing everything a tool-calling LLM needs to discovery/share/find/manage data and tools for itself and in conjunction with other agents. But it's also a library or service you can use yourself to implement a generic CRUD server for protobufs and grpc with powerful, dynamic distributed programming/reflection.

Most of the rest of the docs were written by robots, but reviewed by humans with love in San Francisco.

Architecture

Single Collector

Each collector runs one gRPC server with all services registered:

┌─────────────────────────────────────────┐
│         Collector Instance              │
│                                         │
│  ┌───────────────────────────────────┐ │
│  │    Single gRPC Server             │ │
│  │    (port 50051)                   │ │
│  │                                   │ │
│  │  ├─ CollectorRegistry            │ │
│  │  ├─ CollectionService            │ │
│  │  ├─ CollectiveDispatcher         │ │
│  │  └─ CollectionRepo                │ │
│  │                                   │ │
│  │  Registry Validation: ENABLED    │ │
│  └───────────────────────────────────┘ │
└─────────────────────────────────────────┘

Multi-Collector Cluster

Multiple collectors connect to form a distributed system:

┌──────────────────┐         ┌──────────────────┐
│  Collector 1     │◄───────►│  Collector 2     │
│  localhost:50051 │         │  localhost:50052 │
│                  │         │                  │
│  All 4 services  │         │  All 4 services  │
│  With validation │         │  With validation │
└──────────────────┘         └──────────────────┘
        ▲                            ▲
        │                            │
        └──────────┬─────────────────┘
                   │
              Dispatcher
              connects and
              routes between

Service-to-Service Communication

Services communicate via gRPC loopback even when co-located:

┌─────────────────────────────────────────────────┐
│          Single gRPC Server (port 50051)        │
│                                                 │
│  ┌──────────────┐         ┌──────────────┐    │
│  │  Dispatcher  │ ─────>  │   Registry   │    │
│  │              │  gRPC   │              │    │
│  └──────────────┘  call   └──────────────┘    │
│         │              via loopback             │
│         └──────────────────┐                   │
│                            ▼                   │
│                    localhost:50051             │
└─────────────────────────────────────────────────┘
                            │
                            │ (actual gRPC call)
                            ▼
                    gRPC validation interceptor
                            │
                            ▼
                    Registry.ValidateMethod()

Why loopback? - ✅ Validates server wiring (ensures all services properly registered) - ✅ Consistent behavior (same code path as remote calls) - ✅ Full gRPC features (interceptors, middleware, error handling) - ✅ Type safety (registry validation applies)

Core Services

1. CollectorRegistry

Purpose: Centralized service and type registry

Capabilities: - Register protobuf message types and gRPC services - Validate RPC calls against registered types - Dynamic service discovery and lookup - Namespace-based isolation

Key RPCs: - RegisterProto / RegisterService - Register types - LookupService / ValidateMethod - Query registry - ListServices - Discover available services

Documentation: pkg/registry/README.md

2. CollectionService

Purpose: ORM-like storage for protobuf messages

Capabilities: - CRUD operations (Create, Get, Update, Delete, List) - Full-text search (SQLite FTS5) - JSON filtering for complex queries - File attachments (hierarchical file storage) - Custom RPC handlers - Batch operations

Key RPCs: - Create / Get / Update / Delete / List - CRUD - Search - Full-text + JSON queries - Invoke - Custom method execution - Batch - Multi-operation transactions

Documentation: pkg/collection/README.md

3. CollectiveDispatcher

Purpose: Distributed RPC routing

Capabilities: - Connect collectors into a mesh network - Route requests to appropriate collector - Execute service methods locally or remotely - Namespace-aware routing - Registry-validated execution

Key RPCs: - Connect - Establish collector-to-collector links - Serve - Execute local service methods - Dispatch - Smart request routing (local or remote)

Documentation: pkg/dispatch/README.md

4. CollectionRepo

Purpose: Multi-collection management

Capabilities: - Create collections dynamically - Discover collections by namespace, message type, or labels - Route requests to appropriate collection - Search across multiple collections - 🆕 Backup and restore collections (point-in-time snapshots) - 🆕 Clone collections (local and remote replication) - 🆕 Fetch collections (pull from remote collectors)

Key RPCs: - CreateCollection - Create new collection - 🆕 DeleteCollection - Delete collection and all data - Discover - Find collections - Route - Get collection endpoint - SearchCollections - Cross-collection search - 🆕 BackupCollection - Create point-in-time backup - 🆕 RestoreBackup - Restore from backup - 🆕 ListBackups / DeleteBackup / VerifyBackup - Backup management - 🆕 Clone - Clone collection (local or remote) - 🆕 Fetch - Pull collection from remote collector

Documentation: - pkg/collection/README.md - 🆕 Backup API Guide - Complete backup documentation - 🆕 Clone & Fetch Guide - Replication and migration

Using Collector as a Library

Collector can be easily embedded in your own Go applications. Instead of copying boilerplate code from main.go, use the pkg/server package:

package main

import (
    "log"
    "github.com/accretional/collector/pkg/server"
)

func main() {
    // Create and start the Collector server
    srv, err := server.New(server.Config{
        DataDir:   "./my-app-data",
        Port:      8080,
        Namespace: "my-app",
        // CollectorID auto-generates a UUID7 if not specified
    })
    if err != nil {
        log.Fatalf("Failed to create server: %v", err)
    }
    defer srv.Close()

    // Server is now running with all services available:
    // - CollectorRegistry
    // - CollectionService
    // - CollectiveDispatcher
    // - CollectionRepo

    // Your application logic here...
    // Connect to srv.Address() to use the services

    // Wait for shutdown signal (Ctrl+C)
    srv.WaitForShutdown()
}

Configuration Options: - DataDir - Root directory for all data storage (default: "./data") - Port - gRPC server port (default: 50051) - Namespace - Default namespace for this collector (default: "shared") - CollectorID - Unique identifier for this collector (default: random UUID7) - Logger - Custom logger (default: log.Default())

Reserved Namespaces: - repo, backups, files - Reserved (conflict with filesystem paths) - system - Used for internal collections (types, collections, connections, audit, logs) but not blocked - Use your own namespace for application data (e.g., "my-app", "shared", "production")

See also: examples/embedded/main.go for a complete example.

Quick Start

Running a Collector

# Run the server
go run ./cmd/server/main.go

Output:

Starting Collector (ID: collector-001, Namespace: production)
✓ Registry server created
✓ Registered CollectionService in namespace 'production'
✓ Registered CollectiveDispatcher in namespace 'production'
✓ Registered CollectionRepo in namespace 'production'
✓ Collection repository created
✓ Dispatcher created with gRPC-based registry validation

========================================
Collector collector-001 running on localhost:50051
All services available:
  - CollectorRegistry
  - CollectionService
  - CollectiveDispatcher
  - CollectionRepo
Namespace: production
Registry validation: ENABLED
========================================
Press Ctrl+C to shutdown

Client Example

package main

import (
    "context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    pb "github.com/accretional/collector/gen/collector"
)

func main() {
    // Connect to collector
    conn, _ := grpc.NewClient("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
    defer conn.Close()

    ctx := context.Background()

    // 1. Create a collection
    repoClient := pb.NewCollectionRepoClient(conn)
    createResp, _ := repoClient.CreateCollection(ctx, &pb.CreateCollectionRequest{
        Collection: &pb.Collection{
            Namespace: "production",
            Name:      "users",
            MessageType: &pb.MessageTypeRef{
                Namespace:   "myapp",
                MessageName: "User",
            },
        },
    })

    // 2. Insert a record
    collectionClient := pb.NewCollectionServiceClient(conn)
    user := &pb.User{Id: "user-123", Name: "Alice", Email: "alice@example.com"}
    createResp, _ := collectionClient.Create(ctx, &pb.CreateRequest{
        Collection: &pb.Collection{Namespace: "production", Name: "users"},
        Record:     &pb.Record{Id: "user-123", Data: marshalToAny(user)},
    })

    // 3. Search records
    searchResp, _ := collectionClient.Search(ctx, &pb.SearchRequest{
        Collection: &pb.Collection{Namespace: "production", Name: "users"},
        Query:      "alice",
        Limit:      10,
    })

    // 4. Connect to another collector
    dispatcherClient := pb.NewCollectiveDispatcherClient(conn)
    connectResp, _ := dispatcherClient.Connect(ctx, &pb.ConnectRequest{
        Address:    "localhost:50052",
        Namespaces: []string{"production"},
        Metadata: map[string]string{
            "collector_id": "collector-001",
        },
    })

    // 5. Dispatch a request (routes automatically)
    dispatchResp, _ := dispatcherClient.Dispatch(ctx, &pb.DispatchRequest{
        Namespace:  "production",
        Service:    &pb.ServiceTypeRef{ServiceName: "CollectionService"},
        MethodName: "Get",
        Input:      getRequestAny,
    })
}

Key Features

Namespace-Based Isolation

Everything in Collector is namespaced: - **M