EventStreamer: Mismatch between documentation (Push) and implementation (Pull)

[Kamysek]

Hello :)

I am currently implementing a custom EventStreamer using Encore PubSub, and I've run into a significant architectural discrepancy between the documentation and the actual Go interface.

The documentation describes a Push-based model. The consumer registers a callback function, allowing the streamer to "push" events as they arrive:

func (c *Consumer) Consume(ctx context.Context, fn func(*Event, Ack) error) error

However, the actual code enforces a Pull-based model. The workflow engine actively "pulls" events on a receive method:

func (r *EventReceiver) Recv(ctx context.Context) (*Event, Ack, error)

The Encore PubSub is natively Push-based. If the interface changes to the Push-based model the integration would be very trivial where I simply have to map incoming events to the workflow callback.

With the current interface things are a bit more complicated.

Is the library transitioning to the Push model described in the documentation? If so, is there a specific branch or tag I should target?

Greetings :)

[andrewwormald]

Hey @Kamysek 👋 Thanks for the detailed report and for trying to integrate with Encore PubSub. I think this is a really cool adapter that you are building and would love to take a look when you're done as I think it would be useful for other users as Encore has grown over the years.

You're absolutely correct - there is a discrepancy between the documentation and the actual code. As you suspected, the documentation is incorrect (it was generated by AI agents in an attempt to quickly cover a lot of ground).

The actual implementation uses a pull-based model and this is the correct design. The current interface is:

type EventStreamer interface {
    NewSender(ctx context.Context, topic string) (EventSender, error)
    NewReceiver(ctx context.Context, topic string, name string, opts ...ReceiverOption) (EventReceiver, error)
}

type EventReceiver interface {
    Recv(ctx context.Context) (*Event, Ack, error)
    Close() error
}

Unfortunately, I understand this makes integration with push-based systems like Encore PubSub a bit more involved. You'll need to bridge the push and pull models, what I recommend is using a buffered channel or some form of wrapped in-memory pool that can be pulled from. So something potentially like this:

```go
type encorePubSubReceiver struct {
     eventChan chan *eventWithAck
     closeCh   chan struct{}
}

type eventWithAck struct {
     event *workflow.Event
     ack   workflow.Ack
}

func (r *encorePubSubReceiver) handlePubSubMessage(msg *PubSubMessage) error {
     event := convertToWorkflowEvent(msg)

     r.eventChan <- &eventWithAck{event: event, ack: msg.Ack}
     return nil
}

func (r *encorePubSubReceiver) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error) {
     select {
     case ea := <-r.eventChan:
         return ea.event, ea.ack, nil
     case <-r.closeCh:
         return nil, nil, errors.New("receiver closed")
     case <-ctx.Done():
         return nil, nil, ctx.Err()
     }
}

I'll update the documentation to reflect the actual interface. Thanks again for catching this!

[andrewwormald]

@Kamysek feel free to close the issue if you feel the answer above clarifies things otherwise feel free to ask any questions that you may have.

[Kamysek]

Hi @andrewwormald, I resolved the issue by implementing the existing interface found in the codebase. However, I wanted to double-check if the interface is stable or if it will change in the future since the documentation was different?

I’ve shared my implementation below. I’d appreciate any feedback you might have!

Migration

CREATE TABLE workflow_inbox (
    id BIGSERIAL PRIMARY KEY,
    topic TEXT NOT NULL,
    foreign_id TEXT NOT NULL,
    status_type INTEGER NOT NULL,
    headers JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),

    visible_until TIMESTAMP WITH TIME ZONE
);

CREATE INDEX idx_workflow_inbox_topic_visible_until_created ON workflow_inbox (topic, visible_until, created_at);

Queries

-- name: EnqueueEvent :exec
INSERT INTO workflow_inbox (
    topic, foreign_id, status_type, headers
) VALUES (
    sqlc.arg('topic'), sqlc.arg('foreign_id'), sqlc.arg('status_type'), sqlc.arg('headers')
);

-- name: GetNextEvent :one
UPDATE workflow_inbox
SET visible_until = sqlc.arg('visible_until')
WHERE id = (
    SELECT id
    FROM workflow_inbox AS wi
    WHERE wi.topic = sqlc.arg('topic')
      AND (wi.visible_until IS NULL OR wi.visible_until < NOW())
    ORDER BY wi.created_at ASC
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
RETURNING *;

-- name: DeleteEvent :exec
DELETE FROM workflow_inbox WHERE id = sqlc.arg('id');

Streamer Implementation with Encore PubSub

package encorestreamer

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"encore.dev/pubsub"
	"encore.dev/rlog"
	"encore.dev/storage/sqldb"
	"github.com/luno/workflow"

	db "encore.app/workflow/database/gen"
)

var (
	wdb = sqldb.Named("workflow")
	q   = db.New(wdb.Stdlib())
)

type EventPayload struct {
	Topic      string            `json:"topic"`
	ForeignID  string            `json:"foreign_id"`
	StatusType int               `json:"status_type"`
	Headers    map[string]string `json:"headers"`
	CreatedAt  time.Time         `json:"created_at"`
}

var WorkflowEvents = pubsub.NewTopic[*EventPayload]("workflow-events", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

var _ = pubsub.NewSubscription(WorkflowEvents, "inbox-writer", pubsub.SubscriptionConfig[*EventPayload]{
	Handler: func(ctx context.Context, event *EventPayload) error {
		// Filter out run-state-change events from the inbox
		if strings.HasSuffix(event.Topic, "-run-state-change") {
			return nil
		}

		headersJSON, err := json.Marshal(event.Headers)
		if err != nil {
			rlog.Error("failed to marshal event headers", "err", err, "topic", event.Topic, "foreign_id", event.ForeignID)
			return err
		}

		return q.EnqueueEvent(ctx, &db.EnqueueEventParams{
			Topic:      event.Topic,
			ForeignID:  event.ForeignID,
			StatusType: int(event.StatusType),
			Headers:    headersJSON,
		})
	},
})

type Streamer struct{}

func New() *Streamer {
	return &Streamer{}
}

var _ workflow.EventStreamer = (*Streamer)(nil)

func (s *Streamer) NewSender(ctx context.Context, topic string) (workflow.EventSender, error) {
	return &Sender{topic: topic}, nil
}

func (s *Streamer) NewReceiver(ctx context.Context, topic string, name string, opts ...workflow.ReceiverOption) (workflow.EventReceiver, error) {
	var options workflow.ReceiverOptions
	for _, opt := range opts {
		opt(&options)
	}

	return &Receiver{
		topic:   topic,
		options: options,
	}, nil
}

type Sender struct {
	topic string
}

func (s *Sender) Send(ctx context.Context, foreignID string, statusType int, headers map[workflow.Header]string) error {
	headerMap := make(map[string]string)
	for k, v := range headers {
		headerMap[string(k)] = v
	}

	_, err := WorkflowEvents.Publish(ctx, &EventPayload{
		Topic:      s.topic,
		ForeignID:  foreignID,
		StatusType: statusType,
		Headers:    headerMap,
		CreatedAt:  time.Now(),
	})
	return err
}

func (s *Sender) Close() error {
	return nil
}

type Receiver struct {
	topic   string
	options workflow.ReceiverOptions
}

func (r *Receiver) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error) {
	pollFreq := 500 * time.Millisecond
	if r.options.PollFrequency > 0 {
		pollFreq = r.options.PollFrequency
	}

	maxSleep := 10 * time.Second
	if pollFreq > maxSleep {
		maxSleep = pollFreq
	}
	currentSleep := pollFreq

	for {
		select {
		case <-ctx.Done():
			return nil, nil, ctx.Err()
		case <-time.After(currentSleep):
			// Attempt to fetch the next event from Postgres
			// We give it a 5-minute visibility timeout. If Ack isn't called by then,
			// another worker can pick it up.
			deadline := time.Now().Add(5 * time.Minute)
			eventRow, err := q.GetNextEvent(ctx, &db.GetNextEventParams{
				Topic:        r.topic,
				VisibleUntil: &deadline,
			})

			if err != nil {
				if errors.Is(err, sql.ErrNoRows) {
					// Queue empty, wait longer next time up to maxSleep
					currentSleep *= 2
					if currentSleep > maxSleep {
						currentSleep = maxSleep
					}
					continue
				}
				return nil, nil, fmt.Errorf("db polling error: %w", err)
			}

			var headers map[workflow.Header]string
			if err := json.Unmarshal(eventRow.Headers, &headers); err != nil {
				rlog.Error("failed to unmarshal event headers", "err", err, "topic", r.topic, "foreign_id", eventRow.ForeignID)
				return nil, nil, fmt.Errorf("unmarshal headers: %w", err)
			}

			var createdAt time.Time
			if eventRow.CreatedAt != nil {
				createdAt = *eventRow.CreatedAt
			}

			wEvent := &workflow.Event{
				ID:        int64(eventRow.ID),
				ForeignID: eventRow.ForeignID,
				Type:      int(eventRow.StatusType),
				Headers:   headers,
				CreatedAt: createdAt,
			}

			ack := func() error {
				ackCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
				defer cancel()

				return q.DeleteEvent(ackCtx, eventRow.ID)
			}

			return wEvent, ack, nil
		}
	}
}

func (r *Receiver) Close() error {
	return nil
}