Add workflowstreams contrib package

.github/CODEOWNERS

@@ -2,3 +2,4 @@
 # https://docs.github.com/en/github/creating-cloning-and-archiving-repositories/about-code-owners#codeowners-syntax
 
 * @temporalio/sdk
+/contrib/workflowstreams/ @temporalio/sdk @temporalio/ai-sdk

contrib/workflowstreams/README.md

@@ -0,0 +1,128 @@
+# Workflow Streams
+
+A durable publish/subscribe log hosted inside a Temporal Workflow.
+
+External code (activities, starters, other workflows) publishes messages to
+named topics via **signals**; subscribers long-poll for new items via
+**updates**; a **query** exposes the current offset. The stream is backed by
+Temporal's durable execution, giving ordered, durable, exactly-once delivery
+with client-side batching, publisher dedup, continue-as-new survival,
+truncation, and ~1 MB response paging.
+
+It is well suited to durable event streams whose cost scales with durable
+batches rather than message count. Each poll round-trip costs ~100 ms of
+latency, so it is not intended for ultra-low-latency streaming.
+
+## Workflow side
+
+Construct a `WorkflowStream` once at the start of your workflow. The constructor
+registers the publish signal, poll update, and offset query handlers.
+
+```go
+func MyWorkflow(ctx workflow.Context, priorState *workflowstreams.WorkflowStreamState) error {
+	stream, err := workflowstreams.NewStream(ctx, priorState)
+	if err != nil {
+		return err
+	}
+
+	// Optionally publish from workflow code:
+	if err := stream.Topic("events").Publish("hello from the workflow"); err != nil {
+		return err
+	}
+
+	// Run your workflow; the stream serves external publishers and subscribers
+	// for as long as the workflow is running.
+	return workflow.Await(ctx, func() bool { return done })
+}
+```
+
+For workflows that support continue-as-new, thread a
+`*workflowstreams.WorkflowStreamState` field through your workflow input and
+pass it to `NewStream` — it is `nil` on a fresh start and carries accumulated
+state across continue-as-new boundaries:
+
+```go
+return stream.ContinueAsNew(ctx, MyWorkflow, func(state *workflowstreams.WorkflowStreamState) []any {
+	return []any{state}
+})
+```
+
+## Publishing (client side)
+
+From an activity, use `NewClientFromActivity` to target the parent workflow:
+
+```go
+func PublishActivity(ctx context.Context) error {
+	c, err := workflowstreams.NewClientFromActivity(ctx, workflowstreams.Options{})
+	if err != nil {
+		return err
+	}
+	defer c.Close(ctx)
+
+	topic := c.Topic("events")
+	for i := range 100 {
+		topic.Publish(fmt.Sprintf("item %d", i), false)
+	}
+	return nil // Close flushes the remaining buffer
+}
+```
+
+From a starter or any code with a `client.Client`, use `NewClient` with an
+explicit workflow ID:
+
+```go
+c := workflowstreams.NewClient(temporalClient, workflowID, workflowstreams.Options{})
+defer c.Close(ctx)
+c.Topic("events").Publish("from outside", true /* forceFlush */)
+```
+
+Items are buffered and flushed automatically every `BatchInterval` (default 2s),
+when the buffer reaches `MaxBatchSize`, on `forceFlush`, on an explicit
+`Flush`, or on `Close`.
+
+## Subscribing
+
+`Subscribe` returns a range-over-func iterator (Go 1.23+):
+
+```go
+for item, err := range c.Subscribe(ctx, workflowstreams.SubscribeOptions{
+	Topics: []string{"events"}, // empty/nil = all topics
+}) {
+	if err != nil {
+		return err
+	}
+	var s string
+	if err := converter.GetDefaultDataConverter().FromPayload(item.Data, &s); err != nil {
+		return err
+	}
+	fmt.Printf("offset=%d topic=%s value=%s\n", item.Offset, item.Topic, s)
+}
+```
+
+Breaking out of the loop or cancelling `ctx` stops the subscription and tears
+down the poll loop. The iterator ends cleanly when the workflow reaches a
+terminal state, automatically follows continue-as-new chains, and recovers from
+truncation by restarting from the current base offset.
+
+Items yield the raw `*commonpb.Payload`; decode at the call site with your data
+converter. Offsets are **global** (across all topics), not per-topic.
+
+## Options
+
+| Option | Default | Meaning |
+| --- | --- | --- |
+| `BatchInterval` | 2s | Automatic flush interval |
+| `MaxBatchSize` | unset | Flush once the buffer reaches this size |
+| `MaxRetryDuration` | 10m | Max time to retry a failed flush before `FlushTimeoutError`. Must be < the workflow's publisher TTL (15m) to preserve exactly-once delivery |
+| `DataConverter` | default | Converter for publishing/decoding |
+| `SubscribeOptions.PollCooldown` | 100ms | Min interval between polls |
+
+## Cross-language protocol
+
+The handler names (`PublishSignalName`, `PollUpdateName`, `OffsetQueryName`),
+the JSON envelope field names, and the per-item payload encoding (base64 of the
+marshaled `temporal.api.common.v1.Payload`) match the Python and TypeScript
+packages exactly, so a Go publisher or subscriber interoperates with a
+Python/TypeScript workflow and vice versa. The data converter codec chain
+(encryption, compression) runs once on the signal/update envelope — never per
+item — so payloads are not double-encoded.

contrib/workflowstreams/client.go

@@ -0,0 +1,242 @@
+package workflowstreams
+
+import (
+	"context"
+	"errors"
+	"iter"
+	"sync"
+	"time"
+
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/sdk/activity"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/converter"
+	"go.temporal.io/sdk/temporal"
+)
+
+// Options configures a Client.
+type Options struct {
+	// BatchInterval is the interval between automatic flushes. Default: 2s.
+	BatchInterval time.Duration
+	// MaxBatchSize triggers a flush once the buffer reaches this many items.
+	// Zero disables size-based flushing.
+	MaxBatchSize int
+	// MaxRetryDuration is the maximum time to retry a failed flush before
+	// returning a FlushTimeoutError. Must be less than the workflow's publisher
+	// TTL (default 15m) to preserve exactly-once delivery. Default: 10m.
+	MaxRetryDuration time.Duration
+	// DataConverter is used to turn published values into Payloads and is the
+	// converter callers should use to decode subscribed items. Default:
+	// converter.GetDefaultDataConverter().
+	DataConverter converter.DataConverter
+}
+
+// SubscribeOptions configures a subscription.
+type SubscribeOptions struct {
+	// Topics filters the subscription. Empty or nil means all topics.
+	Topics []string
+	// FromOffset is the global offset to start from. Zero means the beginning.
+	FromOffset int64
+	// PollCooldown is the minimum interval between polls when no more items are
+	// immediately ready. Default: 100ms.
+	PollCooldown time.Duration
+}
+
+// Client publishes to and subscribes from a workflow stream from external code
+// (activities, starters, other workflows). The publish path is owned by an
+// internal publisher; the Client itself holds the target workflow and the read
+// (subscribe/query) surface.
+type Client struct {
+	c          client.Client
+	workflowID string
+	followCAN  bool
+	pub        *publisher
+
+	mu           sync.Mutex
+	topicHandles map[string]*TopicHandle
+}
+
+// NewClient creates a Client targeting workflowID through the given Temporal
+// client. The returned Client follows continue-as-new chains in Subscribe.
+func NewClient(c client.Client, workflowID string, opts Options) *Client {
+	dc := opts.DataConverter
+	if dc == nil {
+		dc = converter.GetDefaultDataConverter()
+	}
+	wsc := &Client{
+		c:            c,
+		workflowID:   workflowID,
+		followCAN:    true,
+		topicHandles: map[string]*TopicHandle{},
+	}
+	wsc.pub = newPublisher(func(ctx context.Context, in PublishInput) error {
+		return c.SignalWorkflow(ctx, workflowID, "", PublishSignalName, in)
+	}, dc, opts)
+	return wsc
+}
+
+// NewClientFromActivity creates a Client targeting the current activity's parent
+// workflow, using the activity's Temporal client. It returns an error if the
+// activity has no parent workflow (a standalone activity); in that case use
+// NewClient with an explicit workflow ID.
+func NewClientFromActivity(ctx context.Context, opts Options) (*Client, error) {
+	info := activity.GetInfo(ctx)
+	if info.WorkflowExecution.ID == "" {
+		return nil, errors.New("workflowstreams: NewClientFromActivity requires an activity scheduled by a workflow; " +
+			"from a standalone activity use NewClient with an explicit workflow ID")
+	}
+	return NewClient(activity.GetClient(ctx), info.WorkflowExecution.ID, opts), nil
+}
+
+// Topic returns a handle for publishing to and subscribing from name. Repeated
+// calls with the same name return the same handle.
+func (c *Client) Topic(name string) *TopicHandle {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if h, ok := c.topicHandles[name]; ok {
+		return h
+	}
+	h := &TopicHandle{name: name, client: c}
+	c.topicHandles[name] = h
+	return h
+}
+
+// Flush sends buffered (and pending) items and waits for server confirmation.
+// It returns once the items buffered at call time have been signaled to the
+// workflow and acknowledged. Returns a FlushTimeoutError if a pending batch
+// cannot be sent within MaxRetryDuration.
+func (c *Client) Flush(ctx context.Context) error { return c.pub.flush(ctx) }
+
+// Close stops the background publisher and drains any remaining items. Call it
+// (e.g. via defer) to guarantee a final flush. It surfaces any deferred
+// FlushTimeoutError from a prior background flush failure.
+func (c *Client) Close(ctx context.Context) error { return c.pub.close(ctx) }
+
+// GetOffset queries the current global offset of the stream.
+func (c *Client) GetOffset(ctx context.Context) (int64, error) {
+	val, err := c.c.QueryWorkflow(ctx, c.workflowID, "", OffsetQueryName)
+	if err != nil {
+		return 0, err
+	}
+	var n int64
+	if err := val.Get(&n); err != nil {
+		return 0, err
+	}
+	return n, nil
+}
+
+// Subscribe returns a range-over-func iterator that long-polls for new items.
+// Iterate with:
+//
+//	for item, err := range c.Subscribe(ctx, opts) {
+//		if err != nil { ... }
+//		// use item
+//	}
+//
+// Breaking out of the loop or cancelling ctx stops the subscription and tears
+// down the poll loop. Each yielded item carries the raw *commonpb.Payload in
+// Data; decode it with your data converter. The iterator ends cleanly when the
+// workflow reaches a terminal state, and automatically follows continue-as-new
+// chains.
+func (c *Client) Subscribe(ctx context.Context, opts SubscribeOptions) iter.Seq2[WorkflowStreamItem, error] {
+	return func(yield func(WorkflowStreamItem, error) bool) {
+		pollCooldown := opts.PollCooldown
+		if pollCooldown <= 0 {
+			pollCooldown = defaultPollCooldown
+		}
+		topics := opts.Topics
+		if topics == nil {
+			topics = []string{}
+		}
+		offset := opts.FromOffset
+
+		for {
+			if err := ctx.Err(); err != nil {
+				yield(WorkflowStreamItem{}, err)
+				return
+			}
+
+			var result PollResult
+			handle, err := c.c.UpdateWorkflow(ctx, client.UpdateWorkflowOptions{
+				WorkflowID:   c.workflowID,
+				UpdateName:   PollUpdateName,
+				Args:         []any{PollInput{Topics: topics, FromOffset: offset}},
+				WaitForStage: client.WorkflowUpdateStageCompleted,
+			})
+			if err == nil {
+				err = handle.Get(ctx, &result)
+			}
+			if err != nil {
+				var appErr *temporal.ApplicationError
+				if errors.As(err, &appErr) && appErr.Type() == ErrTypeTruncatedOffset {
+					// Fell behind truncation; restart from the beginning of
+					// whatever still exists.
+					offset = 0
+					continue
+				}
+				// The workflow may have continued-as-new or completed between
+				// polls. Follow the chain, exit cleanly on a terminal state,
+				// otherwise surface the error.
+				if followed := c.followContinueAsNew(ctx); followed {
+					continue
+				}
+				if c.isTerminal(ctx) {
+					return
+				}
+				yield(WorkflowStreamItem{}, err)
+				return
+			}
+
+			for _, wi := range result.Items {
+				payload, derr := decodePayloadWire(wi.Data)
+				if derr != nil {
+					yield(WorkflowStreamItem{}, derr)
+					return
+				}
+				if !yield(WorkflowStreamItem{Topic: wi.Topic, Data: payload, Offset: wi.Offset}, nil) {
+					return
+				}
+			}
+			offset = result.NextOffset
+
+			if !result.MoreReady {
+				select {
+				case <-time.After(pollCooldown):
+				case <-ctx.Done():
+					yield(WorkflowStreamItem{}, ctx.Err())
+					return
+				}
+			}
+		}
+	}
+}
+
+// followContinueAsNew reports whether the target workflow continued-as-new.
+// Because polls use an empty run ID they always address the latest run, so the
+// caller only needs to retry.
+func (c *Client) followContinueAsNew(ctx context.Context) bool {
+	if !c.followCAN {
+		return false
+	}
+	desc, err := c.c.DescribeWorkflowExecution(ctx, c.workflowID, "")
+	if err != nil {
+		return false
+	}
+	return desc.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW
+}
+
+func (c *Client) isTerminal(ctx context.Context) bool {
+	desc, err := c.c.DescribeWorkflowExecution(ctx, c.workflowID, "")
+	if err != nil {
+		return false
+	}
+	switch desc.GetWorkflowExecutionInfo().GetStatus() {
+	case enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
+		enumspb.WORKFLOW_EXECUTION_STATUS_FAILED,
+		enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED,
+		enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED,
+		enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT:
+		return true
+	}
+	return false
+}