cohorly

README

Cohorly Go SDK

The official server-side Go SDK for Cohorly, a self-hosted product analytics platform. The API mirrors the official Mixpanel Go SDK, so migrating code is mostly a matter of swapping the import.

• Zero dependencies (stdlib only), Go 1.22+.
• Synchronous, context-based Client for tracking, profiles and aliasing.
• Optional BufferedClient with a background flush loop and the shared Cohorly SDK retry contract.

Installation

go get github.com/cohorly-io/cohorly-go

Quickstart

package main

import (
	"context"

	cohorly "github.com/cohorly-io/cohorly-go"
)

func main() {
	client := cohorly.NewClient("YOUR_PROJECT_TOKEN",
		cohorly.WithAPIHost("http://localhost:4000"))

	ctx := context.Background()
	err := client.Track(ctx, []*cohorly.Event{
		client.NewEvent("signup", "user-1", map[string]any{
			"plan": "premium",
		}),
	})
	if err != nil {
		// handle error
	}
}

The project token comes from your Cohorly dashboard (each project has one). Authentication is sent as the X-Cohorly-Token header on every request.

Sending events

NewEvent stamps the Cohorly default properties on every event:

Property	Value
distinct_id	the distinctID argument (skipped for cohorly.EmptyDistinctID)
time	current unix time in milliseconds
$insert_id	random UUID v4, used by the server for deduplication
$lib	"go"
$lib_version	SDK version

Override the timestamp or insert id when backfilling:

e := client.NewEvent("purchase", "user-1", map[string]any{"amount": 9.99})
e.AddTime(orderTime)
e.AddInsertID("order-1234") // retries of the same order dedupe server-side
err := client.Track(ctx, []*cohorly.Event{e})

Track accepts any number of events and transparently chunks them into requests of at most 500 events (the server's batch cap). Batches are applied atomically by the server: a rejected batch inserted nothing and is safe to retry as-is.

Buffered (background) mode

For high-throughput services, wrap the client in a BufferedClient. Enqueue is non-blocking; a background goroutine flushes batches on an interval (or as soon as a full batch is queued) and implements the Cohorly SDK retry contract:

• 429 / 5xx / network error: queue kept, exponential backoff (base 2s, doubling, cap 10 min, +/-20% jitter), Retry-After honored (capped).
• 413: flush batch size halved (floor 1); a single event that alone exceeds the 1 MB body cap is dropped.
• 400: batch dropped permanently.
• 401 (invalid token): queue kept, maximum backoff.
• Queue capped at 1000 events; the oldest events are dropped first.

buffered := cohorly.NewBufferedClient(client,
	cohorly.WithFlushInterval(10*time.Second), // default
	cohorly.WithFlushBatchSize(50),            // default, max 500
	cohorly.WithQueueCapacity(1000),           // default
)

buffered.Enqueue(buffered.NewEvent("page_view", "user-1", nil))

// On shutdown: stop the loop and flush whatever is left.
if err := buffered.Close(ctx); err != nil {
	// remaining events could not be delivered
}

BufferedClient embeds *Client, so the synchronous methods (Track, PeopleSet, Alias, ...) remain available on it.

Managing user identity

Link an anonymous id to a known user id:

err := client.Alias(ctx, "user-1", "anon-af3c...") // alias, existing distinct_id

Storing user profiles

user := cohorly.NewPeopleProperties("user-1", map[string]any{
	"name": "Ada",
	"plan": "premium",
})
err := client.PeopleSet(ctx, []*cohorly.PeopleProperties{user})

Other profile operations:

client.PeopleSetOnce(ctx, []*cohorly.PeopleProperties{user}) // only if unset
client.PeopleIncrement(ctx, "user-1", map[string]float64{"logins": 1})
client.PeopleUnset(ctx, "user-1", []string{"plan"})
client.PeopleDelete(ctx, "user-1") // deletes the profile, keeps events

Error handling

Non-2xx responses are returned as *cohorly.HTTPError with the status code, response body and parsed Retry-After:

err := client.Track(ctx, events)
if errors.Is(err, cohorly.ErrInvalidToken) {
	// bad project token (HTTP 401)
}
var httpErr *cohorly.HTTPError
if errors.As(err, &httpErr) && httpErr.Status == 429 {
	time.Sleep(httpErr.RetryAfter)
	// safe to retry the same payload
}

Differences from mixpanel-go

• NewClient instead of NewApiClient; WithAPIHost/WithHTTPClient options instead of EuResidency/ProxyApiLocation/HttpClient.
• Track chunks batches larger than 500 instead of erroring (Mixpanel caps at 2000).
• PeopleUnset/PeopleDelete instead of PeopleDeleteProperty/ PeopleDeleteProfile (Cohorly has no ignoreAlias); PeopleIncrement takes map[string]float64.
• No Import, groups, or export APIs (not part of the Cohorly server).
• Adds BufferedClient (mixpanel-go is synchronous only).

See PLAN.md for the full mapping.

Documentation

Overview

Package cohorly is the official server-side Go SDK for Cohorly, a self-hosted product analytics platform. Its API mirrors the official Mixpanel Go SDK: a synchronous, context-based Client for tracking events, updating user profiles and aliasing identities, plus an optional BufferedClient that queues events and flushes them in the background with the shared Cohorly SDK retry contract.

Quickstart:

client := cohorly.NewClient("PROJECT_TOKEN",
	cohorly.WithAPIHost("http://localhost:4000"))

err := client.Track(ctx, []*cohorly.Event{
	client.NewEvent("signup", "user-1", map[string]any{"plan": "pro"}),
})

Index

• Constants
• Variables
• type BufferedClient
  • func NewBufferedClient(client *Client, opts ...BufferedOption) *BufferedClient
  • func (b *BufferedClient) Close(ctx context.Context) error
  • func (b *BufferedClient) Enqueue(events ...*Event)
  • func (b *BufferedClient) Flush(ctx context.Context) error
  • func (b *BufferedClient) Len() int
• type BufferedOption
  • func WithFlushBatchSize(n int) BufferedOption
  • func WithFlushInterval(d time.Duration) BufferedOption
  • func WithQueueCapacity(n int) BufferedOption
• type Client
  • func NewClient(token string, opts ...Option) *Client
  • func (c *Client) Alias(ctx context.Context, alias string, distinctID string) error
  • func (c *Client) NewEvent(name string, distinctID string, properties map[string]any) *Event
  • func (c *Client) PeopleDelete(ctx context.Context, distinctID string) error
  • func (c *Client) PeopleIncrement(ctx context.Context, distinctID string, add map[string]float64) error
  • func (c *Client) PeopleSet(ctx context.Context, people []*PeopleProperties) error
  • func (c *Client) PeopleSetOnce(ctx context.Context, people []*PeopleProperties) error
  • func (c *Client) PeopleUnset(ctx context.Context, distinctID string, unset []string) error
  • func (c *Client) Track(ctx context.Context, events []*Event) error
• type Event
  • func (e *Event) AddInsertID(insertID string)
  • func (e *Event) AddTime(t time.Time)
• type HTTPError
  • func (e *HTTPError) Error() string
  • func (e *HTTPError) Unwrap() error
• type Option
  • func WithAPIHost(host string) Option
  • func WithHTTPClient(httpClient *http.Client) Option
• type PeopleProperties
  • func NewPeopleProperties(distinctID string, properties map[string]any) *PeopleProperties

Constants

const (
	// DefaultFlushInterval is how often the background loop flushes the
	// queue when nothing else wakes it.
	DefaultFlushInterval = 10 * time.Second
	// DefaultFlushBatchSize is the number of events sent per flush request.
	DefaultFlushBatchSize = 50
	// DefaultQueueCapacity is the maximum number of queued events; when
	// full, the oldest events are dropped.
	DefaultQueueCapacity = 1000
)

Buffered client defaults.

const DefaultAPIHost = "http://localhost:4000"

DefaultAPIHost is the API host used when WithAPIHost is not provided.

const EmptyDistinctID = ""

EmptyDistinctID can be passed to NewEvent when the event has no associated user. Note that Cohorly expects a distinct_id on every event; use this only when you set distinct_id yourself via the properties map.

const MaxTrackBatch = 500

MaxTrackBatch is the maximum number of events the Cohorly server accepts in a single /track request. Track transparently chunks larger slices into sequential requests of at most this size.

const Version = "0.1.0"

Version is the SDK version, stamped on every event as $lib_version.

Variables

var ErrInvalidToken = errors.New("cohorly: invalid token")

ErrInvalidToken indicates the server rejected the project token (HTTP 401). It is surfaced via errors.Is on the *HTTPError returned by API calls.

Types

type BufferedClient

type BufferedClient struct {
	*Client
	// contains filtered or unexported fields
}

BufferedClient wraps a Client with an in-memory queue and a background flush loop. Enqueue is non-blocking; events are delivered in batches and, on retryable failures, kept and retried with the shared Cohorly SDK retry contract:

• 429/5xx/network error: keep the queue and retry with exponential backoff (base 2s, doubling, cap 10 min, +/-20% jitter), honoring Retry-After (capped at 10 min).
• 413: halve the flush batch size (floor 1) and retry; a single event that alone exceeds the server body cap is dropped.
• 400 (and other permanent 4xx): drop the batch.
• 401: keep the queue, back off at the maximum delay.
• Queue capped (default 1000 events); the oldest events are dropped first.

Server batch rejections are atomic, so retrying the same payload never duplicates events. A BufferedClient is safe for concurrent use.

func NewBufferedClient

func NewBufferedClient(client *Client, opts ...BufferedOption) *BufferedClient

NewBufferedClient creates a BufferedClient on top of client and starts its background flush loop. Call Close to flush remaining events and stop the loop.

func (*BufferedClient) Close

func (b *BufferedClient) Close(ctx context.Context) error

Close stops the background loop and attempts a final synchronous Flush of any remaining events. After Close the client must not be used.

func (*BufferedClient) Enqueue

func (b *BufferedClient) Enqueue(events ...*Event)

Enqueue adds events to the queue without blocking. When the queue is at capacity the oldest events are dropped. A flush is triggered early once a full batch is queued.

func (*BufferedClient) Flush

func (b *BufferedClient) Flush(ctx context.Context) error

Flush synchronously drains the queue, sending batch after batch. It stops and returns the underlying error on the first retryable failure (the queue is kept for the background loop to retry); permanently rejected batches are dropped as per the retry contract.

func (*BufferedClient) Len

func (b *BufferedClient) Len() int

Len returns the number of events currently queued.

type BufferedOption

type BufferedOption func(*BufferedClient)

BufferedOption configures a BufferedClient.

func WithFlushBatchSize

func WithFlushBatchSize(n int) BufferedOption

WithFlushBatchSize sets how many events are sent per flush request (clamped to MaxTrackBatch). Defaults to DefaultFlushBatchSize.

func WithFlushInterval

func WithFlushInterval(d time.Duration) BufferedOption

WithFlushInterval sets how often the background loop flushes the queue. Defaults to DefaultFlushInterval.

func WithQueueCapacity

func WithQueueCapacity(n int) BufferedOption

WithQueueCapacity sets the maximum number of queued events; when the queue is full the oldest events are dropped. Defaults to DefaultQueueCapacity.

type Client

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

Client is a synchronous Cohorly API client. It is safe for concurrent use.

func NewClient

func NewClient(token string, opts ...Option) *Client

NewClient creates a Cohorly API client authenticated by the given project token.

func (*Client) Alias

func (c *Client) Alias(ctx context.Context, alias string, distinctID string) error

Alias links a new identifier to an existing distinct_id so events tracked under either resolve to the same user.

func (*Client) NewEvent

func (c *Client) NewEvent(name string, distinctID string, properties map[string]any) *Event

NewEvent creates an event with the Cohorly default properties stamped:

• distinct_id: distinctID (skipped when EmptyDistinctID is passed)
• time: current unix time in milliseconds (unless already present in properties; override with AddTime)
• $insert_id: a random UUID v4 for server-side deduplication (unless already present; override with AddInsertID)
• $lib: "go"
• $lib_version: Version

The properties map is copied; the caller's map is not mutated.

func (*Client) PeopleDelete

func (c *Client) PeopleDelete(ctx context.Context, distinctID string) error

PeopleDelete deletes the user profile ($delete). Event history is kept.

func (*Client) PeopleIncrement

func (c *Client) PeopleIncrement(ctx context.Context, distinctID string, add map[string]float64) error

PeopleIncrement adds the given numeric deltas to profile properties ($add).

func (*Client) PeopleSet

func (c *Client) PeopleSet(ctx context.Context, people []*PeopleProperties) error

PeopleSet sets profile properties on the given users ($set), overwriting existing values.

func (*Client) PeopleSetOnce

func (c *Client) PeopleSetOnce(ctx context.Context, people []*PeopleProperties) error

PeopleSetOnce sets profile properties on the given users only if they are not already set ($set_once).

func (*Client) PeopleUnset

func (c *Client) PeopleUnset(ctx context.Context, distinctID string, unset []string) error

PeopleUnset removes the given properties from a profile ($unset).

func (*Client) Track

func (c *Client) Track(ctx context.Context, events []*Event) error

Track sends events to the Cohorly /track endpoint. Slices larger than MaxTrackBatch are chunked into sequential requests; if a chunk fails its error is returned and the remaining chunks are not sent. The server applies batches atomically, so a failed batch inserted nothing and is safe to retry as-is.

type Event

type Event struct {
	// Name is the event name.
	Name string `json:"event"`
	// Properties holds the event properties, including the reserved keys
	// stamped by NewEvent (distinct_id, time, $insert_id, $lib,
	// $lib_version).
	Properties map[string]any `json:"properties"`
}

Event is a single analytics event sent to the /track endpoint.

func (*Event) AddInsertID

func (e *Event) AddInsertID(insertID string)

AddInsertID overrides the $insert_id used for server-side deduplication.

func (*Event) AddTime

func (e *Event) AddTime(t time.Time)

AddTime overrides the event timestamp (stored as unix milliseconds).

type HTTPError

type HTTPError struct {
	// Status is the HTTP status code.
	Status int
	// Body is the raw response body (truncated to 64 KiB).
	Body string
	// RetryAfter is the parsed Retry-After header on 429 responses, or 0
	// when the header is absent.
	RetryAfter time.Duration
}

HTTPError is returned for any non-2xx response from the Cohorly server.

func (*HTTPError) Error

func (e *HTTPError) Error() string

Error implements the error interface.

func (*HTTPError) Unwrap

func (e *HTTPError) Unwrap() error

Unwrap maps 401 responses to ErrInvalidToken so that errors.Is(err, cohorly.ErrInvalidToken) works.

type Option

type Option func(*Client)

Option configures a Client.

func WithAPIHost

func WithAPIHost(host string) Option

WithAPIHost sets the base URL of the Cohorly server, e.g. "https://analytics.example.com". Defaults to DefaultAPIHost.

func WithHTTPClient

func WithHTTPClient(httpClient *http.Client) Option

WithHTTPClient sets the *http.Client used for all requests. Defaults to http.DefaultClient.

type PeopleProperties

type PeopleProperties struct {
	// DistinctID identifies the user profile.
	DistinctID string
	// Properties are the profile properties to write.
	Properties map[string]any
}

PeopleProperties is a profile update for a single user, used by PeopleSet and PeopleSetOnce.

func NewPeopleProperties

func NewPeopleProperties(distinctID string, properties map[string]any) *PeopleProperties

NewPeopleProperties creates a profile update for distinctID.