cohorly

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 10, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

README

Cohorly Go SDK

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

Documentation: Go SDK reference · Quickstart · HTTP API

  • 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("https://cohorly-service.velloalabs.com"))

	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

Note: PeopleUnset/PeopleDelete are destructive and gated server-side: the server refuses them on the project token alone (they need an org-owner or superadmin Authorization credential this SDK does not send) and answers HTTP 200 with {"status": 0, ..., "refused"}. Use the dashboard or the admin privacy API for profile removal.

Feature flags

Server-side evaluation is never cached: every call is a direct request to POST /flags/evaluate, so a flag flipped in the dashboard takes effect on the next call.

if on, err := client.IsFeatureEnabled(ctx, "new-nav", "user-1"); err == nil && on {
	// ...
}

// Variant wins when set: branch on it for multivariate flags, and use
// `enabled` only for boolean flags (variant "").
variant, enabled, err := client.GetFeatureFlag(ctx, "cta-color", "user-1")

// Raw JSON payload of the served variant, nil when there is none.
raw, err := client.GetFeatureFlagPayload(ctx, "cta-color", "user-1")

// Every flag in the project, keyed by flag key.
flags, err := client.GetAllFlags(ctx, "user-1") // map[string]cohorly.FlagResult

An unknown flag is not an error: it evaluates to disabled, no variant, no payload. Only transport and non-2xx responses return an error.

Local evaluation

With a flag secret the client polls GET /flags/local-evaluation in the background (default every 30s) and evaluates flags in-process, at zero request latency. The flag secret is minted in the dashboard, is per-project and revocable, and is distinct from the project token - it authorizes that one endpoint and nothing else.

client := cohorly.NewClient("PROJECT_TOKEN",
	cohorly.WithFlagSecret("FLAG_SECRET"),
	cohorly.WithFlagPollInterval(30*time.Second))
defer client.Close() // stops the background poller

Two things to know (ADR-0011):

  • Local evaluation buckets on the raw distinct id - the identity graph lives on the server. Pass the identified user id, as you would anyway for a server-side SDK.
  • A flag whose rules reference a cohort cannot be evaluated in-process and keeps falling back to POST /flags/evaluate per call. GetAllFlags makes at most one request for those, and on failure returns the locally evaluated flags plus a non-nil error - the remote flags are absent from the map rather than reported disabled, so check the error before reading a missing key as "off".

A failed poll keeps the last definitions: stale targeting beats no targeting. A flag edit takes up to one poll interval to propagate; drop the flag secret when immediacy matters more than latency.

Exposure events

Pass cohorly.WithExposureEvent() to a single-flag read to track a $feature_flag_called event with $feature_flag and $feature_flag_response (the variant, or the enabled state for boolean flags):

on, err := client.IsFeatureEnabled(ctx, "new-nav", "user-1", cohorly.WithExposureEvent())

It is opt-in per call, best effort (a failed send never fails the flag read), and never emitted by GetAllFlags.

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 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("https://cohorly-service.velloalabs.com"))

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

Index

Constants

View Source 
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.

View Source 
const DefaultAPIHost = "https://cohorly-service.velloalabs.com"

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

View Source 
const DefaultFlagPollInterval = 30 * time.Second

DefaultFlagPollInterval is how often flag definitions are refetched when WithFlagPollInterval is not provided.

View Source 
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.

View Source 
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.

View Source 
const Version = "0.2.0"

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

Variables

View Source 
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.

Functions

This section is empty.

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.

With WithFlagSecret the client owns a background definitions poller and must be closed with Close; without it, Close is a no-op and optional.

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) Close added in v0.2.0 

func (c *Client) Close() error

Close stops the background flag-definitions poller and waits for it and any in-flight exposure sends to finish (both are bounded by contexts canceled here, so Close never blocks on an unresponsive host). It is idempotent and safe to call on a client that never started a poller. Close does not invalidate the client: flag reads fall back to remote evaluation afterwards.

func (*Client) GetAllFlags added in v0.2.0 

func (c *Client) GetAllFlags(ctx context.Context, distinctID string) (map[string]FlagResult, error)

GetAllFlags evaluates every flag in the project for distinctID, keyed by flag key. Without a flag secret this is a single /flags/evaluate request. With one it evaluates every locally-evaluable flag in-process and makes at most one request, for the cohort-targeted flags that cannot be evaluated locally.

A failure of that one request is reported as a partial result: the locally evaluated flags are returned along with a non-nil error, and the flags that needed the server are absent from the map rather than reported disabled. Check the error before treating a missing key as "off".

Exposure events are never sent by GetAllFlags.

func (*Client) GetFeatureFlag added in v0.2.0 

func (c *Client) GetFeatureFlag(ctx context.Context, key string, distinctID string, opts ...FlagOption) (variant string, enabled bool, err error)

GetFeatureFlag returns the served variant and enabled state of a flag for distinctID. The variant wins when set: a non-empty variant is the value to branch on, and enabled is only meaningful for boolean flags (variant ""). An unknown flag yields ("", false, nil).

func (*Client) GetFeatureFlagPayload added in v0.2.0 

func (c *Client) GetFeatureFlagPayload(ctx context.Context, key string, distinctID string, opts ...FlagOption) (json.RawMessage, error)

GetFeatureFlagPayload returns the raw JSON payload attached to the flag's served variant, or nil when the flag is unknown or carries no payload. Unmarshal it into whatever shape the payload was authored as.

func (*Client) IsFeatureEnabled added in v0.2.0 

func (c *Client) IsFeatureEnabled(ctx context.Context, key string, distinctID string, opts ...FlagOption) (bool, error)

IsFeatureEnabled reports whether the flag is enabled for distinctID. An unknown flag is reported as disabled with no error; only transport and HTTP failures return an error.

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 FlagDefinition added in v0.2.0 

type FlagDefinition struct {
	Key            string        `json:"key"`
	Name           string        `json:"name"`
	Active         bool          `json:"active"`
	Variants       []FlagVariant `json:"variants"`
	Rules          []FlagRule    `json:"rules"`
	LocalEvaluable bool          `json:"localEvaluable"`
}

FlagDefinition is one flag as served by GET /flags/local-evaluation: the fields an SDK needs to evaluate in-process, and nothing else. LocalEvaluable is false when any rule references a cohort; the SDK must fall back to POST /flags/evaluate for such a flag rather than silently not matching.

type FlagOption added in v0.2.0 

type FlagOption func(*flagCallOptions)

FlagOption customizes a single flag read.

func WithExposureEvent added in v0.2.0 

func WithExposureEvent() FlagOption

WithExposureEvent tracks a "$feature_flag_called" event for this read, so the flag exposure shows up in analysis (funnels, retention) alongside the rest of the project's events. It is opt-in per call: flag reads are often in hot paths where an extra event per call is not wanted.

The event goes through the normal Track path and is best effort - a failure to send it never fails the flag read, and never changes the value returned. GetAllFlags never sends exposure events.

type FlagResult added in v0.2.0 

type FlagResult struct {
	// Enabled reports whether the flag is on for this user.
	Enabled bool `json:"enabled"`
	// Variant is the multivariate value, empty when the flag is a simple
	// boolean rollout.
	Variant string `json:"variant"`
	// Payload is the raw JSON payload attached to the served variant, nil
	// when the flag has none.
	Payload json.RawMessage `json:"payload"`
	// Reason explains the decision (e.g. "rollout", "cohort", "override").
	Reason string `json:"reason"`
}

FlagResult is the evaluation of one feature flag for a distinct id, as returned by POST /flags/evaluate.

type FlagRule added in v0.2.0 

type FlagRule struct {
	// CohortID references a cohort. Its presence makes the whole flag
	// non-locally-evaluable, so a rule carrying it is never evaluated here.
	CohortID *int `json:"cohortId,omitempty"`
	// DistinctIDs is the Override allow-list; nil means the rule applies to
	// everyone.
	DistinctIDs []string `json:"distinctIds,omitempty"`
	RolloutPct  float64  `json:"rolloutPct"`
	// Variant names an existing variant key to serve; empty means the
	// variant is picked by hash.
	Variant string `json:"variant,omitempty"`
}

FlagRule is one targeting rule. Rules are ordered within a flag and the first match wins. DistinctIDs is the Override, an explicit allow-list.

type FlagVariant added in v0.2.0 

type FlagVariant struct {
	Key string `json:"key"`
	// Payload is the raw JSON payload attached to the variant, nil when it
	// has none.
	Payload    json.RawMessage `json:"payload,omitempty"`
	RolloutPct float64         `json:"rolloutPct"`
}

FlagVariant is one variant of a multivariate flag. RolloutPct values across a flag's variants sum to exactly 100 when the list is non-empty.

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 WithFlagPollInterval added in v0.2.0 

func WithFlagPollInterval(d time.Duration) Option

WithFlagPollInterval sets how often flag definitions are refetched. Defaults to DefaultFlagPollInterval. A non-positive interval fetches once and never refreshes. A flag edit takes up to one interval to propagate; remote evaluation (no flag secret) is the option when immediacy matters.

func WithFlagSecret added in v0.2.0 

func WithFlagSecret(secret string) Option

WithFlagSecret enables local flag evaluation (ADR-0011). The flag secret is a per-project revocable credential, distinct from the project token, that authorizes exactly one endpoint: GET /flags/local-evaluation. When it is set the client polls flag definitions in the background and evaluates locally-evaluable flags in-process, at zero request latency. Flags whose rules reference a cohort are served with LocalEvaluable false and keep falling back to POST /flags/evaluate per call.

A client created with this option starts a goroutine; call Close when done.

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.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.