core/server/sandbox/realtime_module.go

// Package sandbox — realtime_module.go
//
//
// Allows extensions to publish events to WebSocket channels.
// Events are scoped to rooms — only clients subscribed to the
// channel receive the event.
//
// Starlark API:
//   realtime.publish(channel, event, data={})
//
// Requires permission: realtime.publish
package sandbox

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"go.starlark.net/starlark"
	"go.starlark.net/starlarkstruct"

	"armature/events"
)

// maxRealtimePayload is the payload size limit (pg_notify safe).
const maxRealtimePayload = 7168 // 7KB

// reservedPrefixes are event label prefixes that extensions cannot publish to.
var reservedPrefixes = []string{
	"user.", "system.", "model.", "role.", "notification.",
	"workflow.", "plugin.", "internal.", "db.", "tool.",
	"trigger.", "extension.", "ping", "pong", "room.",
}

// BuildRealtimeModule creates the "realtime" Starlark module for a package.
// Requires the realtime.publish permission.
func BuildRealtimeModule(ctx context.Context, bus *events.Bus, packageID string) *starlarkstruct.Module {
	return MakeModule("realtime", starlark.StringDict{
		"publish": starlark.NewBuiltin("realtime.publish", realtimePublish(ctx, bus, packageID)),
	})
}

func realtimePublish(ctx context.Context, bus *events.Bus, packageID string) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
	return func(_ *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
		var channel, event string
		var data *starlark.Dict

		if err := starlark.UnpackArgs(b.Name(), args, kwargs,
			"channel", &channel,
			"event", &event,
			"data?", &data,
		); err != nil {
			return nil, err
		}

		if channel == "" {
			return nil, fmt.Errorf("realtime.publish: channel is required")
		}
		if event == "" {
			return nil, fmt.Errorf("realtime.publish: event is required")
		}

		// Reject reserved event prefixes
		for _, prefix := range reservedPrefixes {
			if strings.HasPrefix(event, prefix) {
				return nil, fmt.Errorf("realtime.publish: event %q uses reserved prefix %q", event, prefix)
			}
		}

		// Build payload map
		payload := map[string]any{
			"_pkg": packageID,
		}

		if data != nil {
			for _, item := range data.Items() {
				k, ok := starlark.AsString(item[0])
				if !ok {
					return nil, fmt.Errorf("realtime.publish: dict key must be string, got %s", item[0].Type())
				}
				if k == "_pkg" {
					continue // reserved field
				}
				payload[k] = StarlarkToGo(item[1])
			}
		}

		// Marshal and check size
		jsonBytes, err := json.Marshal(payload)
		if err != nil {
			return nil, fmt.Errorf("realtime.publish: failed to marshal data: %w", err)
		}
		if len(jsonBytes) > maxRealtimePayload {
			return nil, fmt.Errorf("realtime.publish: payload too large (%d bytes, max %d)", len(jsonBytes), maxRealtimePayload)
		}

		// Publish to event bus
		bus.Publish(events.Event{
			Label:   "realtime." + event,
			Room:    channel,
			Payload: jsonBytes,
			Ts:      time.Now().UnixMilli(),
		})

		return starlark.True, nil
	}
}