feat: support ingestion API

Author: emranbm

go.mod

@@ -10,7 +10,10 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
 	golang.org/x/net v0.40.0 // indirect
+	golang.org/x/sys v0.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,15 +1,26 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/divar-ir/go-http-mock v0.0.0-20250513125724-292fce49eb83 h1:38lh5mrH0Jqv3wcy0Vhmd+OUEGFKPvsqh2AdL/YTwOo=
 github.com/divar-ir/go-http-mock v0.0.0-20250513125724-292fce49eb83/go.mod h1:bEFO6+ho24AtbweHj7nhFAfLxHL+Edx9vScCaO/nDgc=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 golang.org/x/net v0.40.0 h1:79Xs7wF06Gbdcg4kdCCIQArK11Z1hr5POQ6+fIYHNuY=
 golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 resty.dev/v3 v3.0.0-beta.3 h1:3kEwzEgCnnS6Ob4Emlk94t+I/gClyoah7SnNi67lt+E=

internal/constants/constants.go

@@ -0,0 +1,13 @@
+package constants
+
+type IngestionEventType string
+
+const (
+	IngestionEventTypeTraceCreate      IngestionEventType = "trace-create"
+	IngestionEventTypeGenerationCreate IngestionEventType = "generation-create"
+	IngestionEventTypeGenerationUpdate IngestionEventType = "generation-update"
+	IngestionEventTypeScoreCreate      IngestionEventType = "score-create"
+	IngestionEventTypeSpanCreate       IngestionEventType = "span-create"
+	IngestionEventTypeSpanUpdate       IngestionEventType = "span-update"
+	IngestionEventTypeEventCreate      IngestionEventType = "event-create"
+)

internal/models/models.go

@@ -0,0 +1,53 @@
+package models
+
+import (
+	"time"
+
+	"github.com/divar-ir/golangfuse/internal/constants"
+)
+
+type Trace struct {
+	ID        string     `json:"id,omitempty"`
+	Timestamp *time.Time `json:"timestamp,omitempty"`
+	Name      string     `json:"name,omitempty"`
+	UserID    string     `json:"userId,omitempty"`
+	Input     any        `json:"input,omitempty"`
+	Output    any        `json:"output,omitempty"`
+	SessionID string     `json:"sessionId,omitempty"`
+	Release   string     `json:"release,omitempty"`
+	Version   string     `json:"version,omitempty"`
+	Metadata  any        `json:"metadata,omitempty"`
+	Tags      []string   `json:"tags,omitempty"`
+	Public    bool       `json:"public,omitempty"`
+}
+
+type Ingestion struct {
+	Batch []IngestionEvent `json:"batch"`
+}
+
+type IngestionEvent struct {
+	ID        string                       `json:"id"`
+	Timestamp time.Time                    `json:"timestamp"`
+	Type      constants.IngestionEventType `json:"type"`
+	Body      any                          `json:"body"`
+}
+
+type PromptItem struct {
+	Role    string `json:"role"`
+	Content string `json:"content"`
+}
+
+type ChatPrompt struct {
+	ID        string       `json:"id"`
+	CreatedAt time.Time    `json:"createdAt"`
+	UpdatedAt time.Time    `json:"updatedAt"`
+	ProjectID string       `json:"projectId"`
+	CreatedBy string       `json:"createdBy"`
+	Prompt    []PromptItem `json:"prompt"`
+	Name      string       `json:"name"`
+	Version   int          `json:"version"`
+	Type      string       `json:"type"`
+	Config    any          `json:"config"`
+	Tags      []string     `json:"tags"`
+	Labels    []string     `json:"labels"`
+}

internal/observer/observer.go

@@ -0,0 +1,47 @@
+package observer
+
+import (
+	"context"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+type Observer[T any] interface {
+	StartObserve(ctx context.Context, period time.Duration)
+}
+
+type observerImpl[T any] struct {
+	queue        Queue[T]
+	eventHandler EventHandler[T]
+}
+
+type EventHandler[T any] func(ctx context.Context, events []T) error
+
+func NewObserver[T any](queue Queue[T], eventHandler EventHandler[T]) Observer[T] {
+	o := &observerImpl[T]{
+		queue:        queue,
+		eventHandler: eventHandler,
+	}
+	return o
+}
+
+func (o *observerImpl[T]) StartObserve(ctx context.Context, period time.Duration) {
+	ticker := time.NewTicker(period)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			items := o.queue.Flush()
+			if len(items) > 0 {
+				err := o.eventHandler(ctx, items)
+				if err != nil {
+					logrus.WithError(err).
+						Error("golangfuse: error in event handler")
+				}
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}

internal/observer/queue.go

@@ -0,0 +1,31 @@
+package observer
+
+import "sync"
+
+type Queue[T any] interface {
+	Enqueue(event T)
+	Flush() []T
+}
+
+type queueImpl[T any] struct {
+	items []T
+	mu    sync.Mutex
+}
+
+func NewQueue[T any]() Queue[T] {
+	return &queueImpl[T]{}
+}
+
+func (q *queueImpl[T]) Enqueue(event T) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.items = append(q.items, event)
+}
+
+func (q *queueImpl[T]) Flush() []T {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	items := q.items
+	q.items = []T{}
+	return items
+}