feat: push Douyin events through shared WebSocket and SSE

internal/controlplane/api/event_channel.go

@@ -0,0 +1,232 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"git.ipao.vip/rogee/creator-hub/internal/creator"
+	"git.ipao.vip/rogee/creator-hub/internal/environment"
+	"github.com/gorilla/websocket"
+)
+
+type channelMessage struct {
+	Type         string                     `json:"type"`
+	Subscription string                     `json:"subscription"`
+	Error        string                     `json:"error"`
+	Deliveries   []creator.ListenerDelivery `json:"deliveries"`
+}
+type eventSubscription struct {
+	hub       *gatewayEventChannel
+	id, alias string
+	messages  chan channelMessage
+}
+type gatewayEventChannel struct {
+	target     environment.Gateway
+	once       sync.Once
+	connectErr error
+	conn       *websocket.Conn
+	write      sync.Mutex
+	mu         sync.Mutex
+	subs       map[string]*eventSubscription
+	done       chan struct{}
+	finish     sync.Once
+}
+
+var eventChannels = struct {
+	sync.Mutex
+	items map[string]*gatewayEventChannel
+}{items: map[string]*gatewayEventChannel{}}
+var subscriptionSequence atomic.Uint64
+
+func acquireEventSubscription(ctx context.Context, target environment.Gateway, params map[string]any) (*eventSubscription, error) {
+	key := target.Endpoint + "\n" + target.Token
+	eventChannels.Lock()
+	hub := eventChannels.items[key]
+	if hub != nil {
+		select {
+		case <-hub.done:
+			hub = nil
+		default:
+		}
+	}
+	if hub == nil {
+		hub = &gatewayEventChannel{target: target, subs: map[string]*eventSubscription{}, done: make(chan struct{})}
+		eventChannels.items[key] = hub
+	}
+	alias, _ := params["alias"].(string)
+	sub := &eventSubscription{hub: hub, id: fmt.Sprint(subscriptionSequence.Add(1)), alias: alias, messages: make(chan channelMessage, 2)}
+	// Reserve the subscription before releasing the pool lock, so closing the
+	// previous last account cannot retire a channel another account is joining.
+	hub.mu.Lock()
+	hub.subs[sub.id] = sub
+	hub.mu.Unlock()
+	eventChannels.Unlock()
+	hub.once.Do(func() {
+		u, err := url.Parse(strings.TrimRight(target.Endpoint, "/") + "/v1/channel")
+		if err != nil {
+			hub.connectErr = err
+			hub.fail(err)
+			return
+		}
+		switch u.Scheme {
+		case "http":
+			u.Scheme = "ws"
+		case "https":
+			u.Scheme = "wss"
+		default:
+			hub.connectErr = errors.New("invalid gateway scheme")
+			hub.fail(hub.connectErr)
+			return
+		}
+		headers := http.Header{"Authorization": []string{"Bearer " + target.Token}}
+		dialer := websocket.Dialer{HandshakeTimeout: 10 * time.Second}
+		hub.conn, _, hub.connectErr = dialer.DialContext(ctx, u.String(), headers)
+		if hub.connectErr != nil {
+			hub.fail(hub.connectErr)
+			return
+		}
+		hub.conn.SetReadLimit(16 << 20)
+		hub.conn.SetPingHandler(func(data string) error {
+			_ = hub.conn.SetReadDeadline(time.Now().Add(45 * time.Second))
+			return hub.conn.WriteControl(websocket.PongMessage, []byte(data), time.Now().Add(5*time.Second))
+		})
+		go hub.receive()
+	})
+	if hub.connectErr != nil {
+		sub.close()
+		return nil, fmt.Errorf("gateway WS connection: %w", hub.connectErr)
+	}
+	value := make(map[string]any, len(params)+2)
+	for k, v := range params {
+		value[k] = v
+	}
+	value["type"] = "subscribe"
+	value["subscription"] = sub.id
+	if err := hub.send(value); err != nil {
+		sub.close()
+		return nil, err
+	}
+	select {
+	case message := <-sub.messages:
+		if message.Type != "subscribed" {
+			sub.close()
+			return nil, fmt.Errorf("gateway WS subscription: %s", message.Error)
+		}
+		return sub, nil
+	case <-hub.done:
+		sub.close()
+		return nil, errors.New("gateway WS disconnected")
+	case <-ctx.Done():
+		sub.close()
+		return nil, ctx.Err()
+	}
+}
+func (h *gatewayEventChannel) send(value any) error {
+	h.write.Lock()
+	defer h.write.Unlock()
+	select {
+	case <-h.done:
+		return errors.New("gateway WS disconnected")
+	default:
+	}
+	if err := h.conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		return err
+	}
+	if err := h.conn.WriteJSON(value); err != nil {
+		h.fail(err)
+		return err
+	}
+	return nil
+}
+func (h *gatewayEventChannel) receive() {
+	for {
+		_ = h.conn.SetReadDeadline(time.Now().Add(45 * time.Second))
+		var message channelMessage
+		if err := h.conn.ReadJSON(&message); err != nil {
+			h.fail(err)
+			return
+		}
+		if message.Type != "subscribed" && message.Type != "deliveries" && message.Type != "error" {
+			h.fail(errors.New("invalid gateway WS message"))
+			return
+		}
+		if message.Type == "error" && message.Subscription == "" {
+			h.fail(errors.New(message.Error))
+			return
+		}
+		h.mu.Lock()
+		sub := h.subs[message.Subscription]
+		h.mu.Unlock()
+		if sub == nil {
+			continue
+		} // A closed generation cannot receive delayed replies.
+		select {
+		case sub.messages <- message:
+		default:
+			h.fail(errors.New("gateway WS subscription queue full"))
+			return
+		}
+	}
+}
+func (h *gatewayEventChannel) fail(err error) {
+	h.finish.Do(func() {
+		close(h.done)
+		if h.conn != nil {
+			_ = h.conn.Close()
+		}
+		h.mu.Lock()
+		defer h.mu.Unlock()
+		for _, sub := range h.subs {
+			select {
+			case sub.messages <- channelMessage{Type: "error", Error: err.Error()}:
+			default:
+			}
+		}
+	})
+}
+func (s *eventSubscription) poll(ctx context.Context, acks []string) ([]creator.ListenerDelivery, error) {
+	if len(acks) > 0 {
+		if err := s.hub.send(map[string]any{"type": "ack", "alias": s.alias, "subscription": s.id, "delivery_ids": acks}); err != nil {
+			return nil, err
+		}
+	}
+	select {
+	case message := <-s.messages:
+		if message.Type != "deliveries" || len(message.Deliveries) == 0 {
+			return nil, fmt.Errorf("gateway event delivery: %s", message.Error)
+		}
+		for _, d := range message.Deliveries {
+			if err := d.Validate(); err != nil {
+				return nil, err
+			}
+		}
+		return message.Deliveries, nil
+	case <-s.hub.done:
+		return nil, errors.New("gateway WS disconnected")
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+func (s *eventSubscription) close() {
+	_ = s.hub.send(map[string]any{"type": "unsubscribe", "alias": s.alias, "subscription": s.id})
+	key := s.hub.target.Endpoint + "\n" + s.hub.target.Token
+	eventChannels.Lock()
+	s.hub.mu.Lock()
+	delete(s.hub.subs, s.id)
+	empty := len(s.hub.subs) == 0
+	if empty && eventChannels.items[key] == s.hub {
+		delete(eventChannels.items, key)
+	}
+	s.hub.mu.Unlock()
+	eventChannels.Unlock()
+	if empty {
+		s.hub.fail(errors.New("gateway event channel stopped"))
+	}
+}

internal/controlplane/api/event_channel_test.go

@@ -0,0 +1,179 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"git.ipao.vip/rogee/creator-hub/internal/environment"
+	"github.com/gorilla/websocket"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestEventChannelSharesConnectionAndAcknowledges(t *testing.T) {
+	var connections atomic.Int32
+	acks := make(chan struct{}, 2)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/channel" || r.Header.Get("Authorization") != "Bearer test" {
+			t.Errorf("unexpected channel request")
+			w.WriteHeader(400)
+			return
+		}
+		c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		defer c.Close()
+		connections.Add(1)
+		for {
+			var v map[string]any
+			if c.ReadJSON(&v) != nil {
+				return
+			}
+			switch v["type"] {
+			case "subscribe":
+				_ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": v["subscription"]})
+				_ = c.WriteJSON(map[string]any{"type": "deliveries", "subscription": v["subscription"], "deliveries": []any{map[string]any{"kind": "open", "delivery_id": v["alias"]}}})
+			case "ack":
+				acks <- struct{}{}
+			}
+		}
+	}))
+	defer server.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	target := environment.Gateway{Endpoint: server.URL, Token: "test"}
+	first, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "a"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer first.close()
+	second, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "b"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer second.close()
+	if connections.Load() != 1 {
+		t.Fatalf("connections=%d", connections.Load())
+	}
+	items, err := first.poll(ctx, nil)
+	if err != nil || items[0].DeliveryID != "a" {
+		t.Fatalf("%+v %v", items, err)
+	}
+	items, err = second.poll(ctx, nil)
+	if err != nil || items[0].DeliveryID != "b" {
+		t.Fatalf("%+v %v", items, err)
+	}
+	short, c := context.WithTimeout(ctx, 30*time.Millisecond)
+	defer c()
+	_, _ = first.poll(short, []string{"a"})
+	select {
+	case <-acks:
+	case <-ctx.Done():
+		t.Fatal("ACK not received")
+	}
+}
+func TestEventChannelRejectsHandshakeAndSubscriptionTimeout(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(503) }))
+	defer server.Close()
+	for _, endpoint := range []string{server.URL, "invalid://gateway", "http://%"} {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		_, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: endpoint}, map[string]any{"alias": "a"})
+		cancel()
+		if err == nil {
+			t.Fatalf("invalid connection accepted: %s", endpoint)
+		}
+	}
+	waiting := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c, e := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+		if e != nil {
+			return
+		}
+		defer c.Close()
+		for {
+			var value map[string]any
+			if c.ReadJSON(&value) != nil {
+				return
+			}
+		}
+	}))
+	defer waiting.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
+	defer cancel()
+	_, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: waiting.URL}, map[string]any{"alias": "a"})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("subscription wait did not honor cancellation: %v", err)
+	}
+}
+
+func TestEventChannelReportsDisconnect(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c, e := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+		if e != nil {
+			return
+		}
+		defer c.Close()
+		var v map[string]any
+		_ = c.ReadJSON(&v)
+		_ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": v["subscription"]})
+	}))
+	defer server.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	sub, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: server.URL}, map[string]any{"alias": "a"})
+	if err == nil {
+		defer sub.close()
+		_, err = sub.poll(ctx, nil)
+	}
+	if err == nil {
+		t.Fatal("disconnect hidden")
+	}
+}
+
+func TestEventChannelClosingLastAccountDoesNotRetireJoiningAccount(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		for {
+			var value map[string]any
+			if c.ReadJSON(&value) != nil {
+				return
+			}
+			if value["type"] == "subscribe" {
+				_ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": value["subscription"]})
+			}
+		}
+	}))
+	defer server.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	target := environment.Gateway{Endpoint: server.URL}
+	previous, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "a"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { previous.close() }()
+	for i := 0; i < 50; i++ {
+		type result struct {
+			sub *eventSubscription
+			err error
+		}
+		joined := make(chan result, 1)
+		go func() {
+			sub, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "b"})
+			joined <- result{sub, err}
+		}()
+		previous.close()
+		next := <-joined
+		if next.err != nil {
+			t.Fatalf("joining account interrupted at iteration %d: %v", i, next.err)
+		}
+		previous = next.sub
+	}
+}

internal/controlplane/api/event_cover_worker.go

@@ -0,0 +1,86 @@
+package api
+
+import (
+	"context"
+	"errors"
+
+	"git.ipao.vip/rogee/creator-hub/internal/creator"
+	"github.com/sirupsen/logrus"
+)
+
+type listenerCoverWriter interface {
+	UpdateListenerEventCovers(context.Context, string, string, []creator.ListenerDelivery) error
+}
+type listenerCoverEnricher interface {
+	enrich(context.Context, []creator.ListenerDelivery) ([]creator.ListenerDelivery, error)
+}
+
+const coverQueueFullReason = "封面待处理队列已满，等待历史核对重试"
+
+// The account receiver is the sole producer: prepare before persistence, submit
+// after commit. Queue overload is recorded in that same transaction, never by
+// a second synchronous database write on the event acknowledgement path.
+func startListenerCoverWorker(ctx context.Context, store listenerWriter, accountID, generation string, session listenerSession) func([]creator.ListenerDelivery) func() {
+	noop := func() {}
+	writer, ok := store.(listenerCoverWriter)
+	enricher, has := session.(listenerCoverEnricher)
+	if !ok || !has {
+		return func([]creator.ListenerDelivery) func() { return noop }
+	}
+	jobs := make(chan []creator.ListenerDelivery, 32)
+	log := logrus.WithFields(logrus.Fields{"account_id": accountID, "generation": generation})
+	save := func(items []creator.ListenerDelivery) {
+		if err := writer.UpdateListenerEventCovers(ctx, accountID, generation, items); err != nil {
+			if !errors.Is(err, creator.ErrConflict) && ctx.Err() == nil {
+				log.WithError(err).Error("persist event cover result failed")
+			}
+			return
+		}
+		publishEventChanges(ctx)
+	}
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case items := <-jobs:
+				result, err := enricher.enrich(ctx, items)
+				if err != nil {
+					log.WithError(err).Error("event cover enrichment failed")
+					for i := range items {
+						items[i].Notice.WorkCoverError = err.Error()
+					}
+					save(items)
+				} else {
+					save(result)
+				}
+			}
+		}
+	}()
+	return func(items []creator.ListenerDelivery) func() {
+		pending := make([]creator.ListenerDelivery, 0, len(items))
+		for _, d := range items {
+			if d.Notice != nil && d.Notice.WorkID != "" && d.Notice.WorkCoverURL != "" {
+				copy := *d.Notice
+				d.Notice = &copy
+				pending = append(pending, d)
+			}
+		}
+		if len(pending) == 0 {
+			return noop
+		}
+		if len(jobs) == cap(jobs) {
+			log.Error("event cover queue full; recording overload with notification persistence")
+			for i, d := range items {
+				if d.Notice != nil && d.Notice.WorkID != "" && d.Notice.WorkCoverURL != "" {
+					copy := *d.Notice
+					copy.WorkCoverError = coverQueueFullReason
+					items[i].Notice = &copy
+				}
+			}
+			return noop
+		}
+		// No other producer can fill this capacity between prepare and commit.
+		return func() { jobs <- pending }
+	}
+}

internal/controlplane/api/event_cover_worker_test.go

@@ -0,0 +1,107 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"git.ipao.vip/rogee/creator-hub/internal/creator"
+	"testing"
+	"time"
+)
+
+type coverWriterStub struct {
+	listenerWriterStub
+	saved chan []creator.ListenerDelivery
+	err   error
+}
+
+func (s *coverWriterStub) UpdateListenerEventCovers(_ context.Context, _, _ string, items []creator.ListenerDelivery) error {
+	s.saved <- items
+	return s.err
+}
+
+type coverEnricherStub struct {
+	listenerSessionStub
+	entered chan struct{}
+	release chan struct{}
+	err     error
+}
+
+func (s *coverEnricherStub) enrich(ctx context.Context, items []creator.ListenerDelivery) ([]creator.ListenerDelivery, error) {
+	select {
+	case s.entered <- struct{}{}:
+	default:
+	}
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-s.release:
+	}
+	for i := range items {
+		items[i].Notice.WorkCoverError = "cached result"
+	}
+	return items, s.err
+}
+func coverDelivery() []creator.ListenerDelivery {
+	return []creator.ListenerDelivery{{Kind: "notice", DeliveryID: "cover", Notice: &creator.ListenerNotice{EventKey: "cover", EventType: "like", WorkID: "9", WorkAuthorUID: "1", WorkCoverURL: "https://example.com/cover"}}}
+}
+func TestEventCoverWorkerDoesNotBlockReceptionAndReportsOverload(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	writer := &coverWriterStub{saved: make(chan []creator.ListenerDelivery, 64)}
+	session := &coverEnricherStub{entered: make(chan struct{}, 1), release: make(chan struct{})}
+	queue := startListenerCoverWorker(ctx, writer, "account", "generation", session)
+	original := coverDelivery()
+	queue(original)()
+	select {
+	case <-session.entered:
+	case <-time.After(time.Second):
+		t.Fatal("cover worker did not start")
+	}
+	for i := 0; i < 32; i++ {
+		queue(coverDelivery())()
+	}
+	overloaded := coverDelivery()
+	queue(overloaded)()
+	if overloaded[0].Notice.WorkCoverError != coverQueueFullReason {
+		t.Fatal("overload must be included in notification persistence")
+	}
+	select {
+	case <-writer.saved:
+		t.Fatal("overload performed a synchronous cover-result write")
+	default:
+	}
+	if original[0].Notice.WorkCoverError != "" {
+		t.Fatal("delivery was mutated before persistence")
+	}
+	close(session.release)
+	select {
+	case saved := <-writer.saved:
+		if saved[0].Notice.WorkCoverError != "cached result" {
+			t.Fatal("result missing")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("cover not saved")
+	}
+}
+func TestEventCoverFailureIsRecordedAndDoesNotChangeListenerStatus(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	release := make(chan struct{})
+	close(release)
+	writer := &coverWriterStub{saved: make(chan []creator.ListenerDelivery, 4), err: creator.ErrConflict}
+	session := &coverEnricherStub{entered: make(chan struct{}, 1), release: release, err: errors.New("cover unavailable")}
+	queue := startListenerCoverWorker(ctx, writer, "account", "generation", session)
+	queue(nil)()
+	queue(coverDelivery())()
+	select {
+	case saved := <-writer.saved:
+		if saved[0].Notice.WorkCoverError != "cover unavailable" {
+			t.Fatal("error hidden")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("error not saved")
+	}
+	if writer.batches != 0 {
+		t.Fatal("cover failure changed listener health")
+	}
+}

internal/controlplane/api/event_listener.go

@@ -28,6 +28,7 @@ func registerAccountEventRoutes(app *fiber.App, store *creator.Store) {
 			return creatorError(c, err)
 		}
 		logrus.WithFields(logrus.Fields{"account_id": state.AccountID, "generation": state.Generation, "enabled": state.Enabled}).Info("account event listener setting changed")
+		creatorUpdates.publish()
 		return c.JSON(state)
 	})
 	app.Get("/api/creator/events", func(c fiber.Ctx) error {

internal/controlplane/api/event_listener_lifecycle_test.go

@@ -5,9 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/gorilla/websocket"
 	"net/http"
 	"net/http/httptest"
-	"net/url"
 	"strings"
 	"sync"
 	"testing"
@@ -131,7 +131,7 @@ func TestGatewayListenerSessionProtocol(t *testing.T) {
 		case http.MethodDelete:
 			w.WriteHeader(http.StatusNoContent)
 		case http.MethodPost:
-			w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123"}`))
+			w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123","session_id":"test-session"}`))
 		case http.MethodGet:
 			w.Write([]byte(`[{"delivery_id":"1","kind":"open"}]`))
 		default:
@@ -144,22 +144,13 @@ func TestGatewayListenerSessionProtocol(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if items, err := session.Start(context.Background()); err != nil || len(items) != 0 {
+	if items, err := session.request(context.Background(), "start"); err != nil || len(items) != 0 {
 		t.Fatalf("start=%+v %v", items, err)
 	}
 	start := <-calls
 	if start["method"] != http.MethodPost || start["expected_uid"] != "123" || start["binding_version"] != float64(7) || start["runtime_id"] != "runtime" {
 		t.Fatalf("start=%+v", start)
 	}
-	session.pollInterval = 0
-	if _, err := session.Poll(context.Background(), []string{"1"}); err != nil {
-		t.Fatal(err)
-	}
-	poll := <-calls
-	query := poll["query"].(url.Values)
-	if poll["method"] != http.MethodGet || len(query["ack"]) != 1 || query.Get("ack") != "1" || query.Get("wait") != "0" || query.Get("limit") != "100" {
-		t.Fatalf("poll=%+v", poll)
-	}
 	if err := session.Stop(context.Background()); err != nil {
 		t.Fatal(err)
 	}
@@ -167,7 +158,7 @@ func TestGatewayListenerSessionProtocol(t *testing.T) {
 	if stop["method"] != http.MethodDelete {
 		t.Fatalf("stop=%+v", stop)
 	}
-	for i := 0; i < 3; i++ {
+	for i := 0; i < 2; i++ {
 		select {
 		case aliases := <-accounts.lockCalls:
 			if len(aliases) != 1 || aliases[0] != "browser" {
@@ -285,6 +276,28 @@ func TestListenerReconcilerHonorsEnableDisableAndRestart(t *testing.T) {
 	var starts, stops int
 	var pending string
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/channel" {
+			c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			defer c.Close()
+			var value map[string]any
+			if c.ReadJSON(&value) != nil {
+				return
+			}
+			_ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": value["subscription"]})
+			mu.Lock()
+			delivery := pending
+			mu.Unlock()
+			_ = c.WriteJSON(map[string]any{"type": "deliveries", "subscription": value["subscription"], "deliveries": []creator.ListenerDelivery{{Kind: "open", DeliveryID: delivery}}})
+			for {
+				if c.ReadJSON(&value) != nil {
+					return
+				}
+			}
+		}
 		w.Header().Set("Content-Type", "application/json")
 		switch {
 		case r.Method == http.MethodPost:
@@ -292,7 +305,7 @@ func TestListenerReconcilerHonorsEnableDisableAndRestart(t *testing.T) {
 			starts++
 			pending = fmt.Sprintf("open-%d", starts)
 			mu.Unlock()
-			w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123"}`))
+			w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123","session_id":"test-session"}`))
 		case r.Method == http.MethodDelete:
 			mu.Lock()
 			stops++

internal/controlplane/api/event_listener_worker.go

@@ -39,14 +39,17 @@ type gatewayListenerSession struct {
 	target       hub.Gateway
 	environment  hub.EnvironmentContext
 	uid          string
+	generation   string
+	sessionID    string
+	enabled      bool
 	resources    browserAliasResourceLocker
 	boundaryAt   *time.Time
 	checkpoints  map[string]string
-	pollInterval time.Duration
 	coverFiles   eventCoverFiles
+	subscription *eventSubscription
 }
 
-func (s gatewayListenerSession) request(ctx context.Context, action string, ack []string) ([]creator.ListenerDelivery, error) {
+func (s *gatewayListenerSession) request(ctx context.Context, action string) ([]creator.ListenerDelivery, error) {
 	if s.resources == nil {
 		return nil, errors.New("event listener browser alias resource locker unavailable")
 	}
@@ -57,6 +60,11 @@ func (s gatewayListenerSession) request(ctx context.Context, action string, ack
 	defer unlock()
 
 	payload := gatewayGenerationPayload(s.environment)
+	payload["listener_generation"] = s.generation
+	payload["session_id"] = s.sessionID
+	if action == "stop" && !s.enabled {
+		payload["stop_all"] = true
+	}
 	path := "/v1/browsers/" + url.PathEscape(s.environment.Alias) + "/douyin/events"
 	method, expectedStatus := http.MethodPost, http.StatusOK
 	switch action {
@@ -68,13 +76,6 @@ func (s gatewayListenerSession) request(ctx context.Context, action string, ack
 			checkpoints = map[string]string{}
 		}
 		payload["notice_checkpoints"] = checkpoints
-	case "poll":
-		method = http.MethodGet
-		query := url.Values{"limit": {"100"}, "wait": {"0"}}
-		for _, id := range ack {
-			query.Add("ack", id)
-		}
-		path += "?" + query.Encode()
 	case "stop":
 		method, expectedStatus = http.MethodDelete, http.StatusNoContent
 	default:
@@ -95,40 +96,43 @@ func (s gatewayListenerSession) request(ctx context.Context, action string, ack
 			Connected bool   `json:"connected"`
 			Alias     string `json:"alias"`
 			UID       string `json:"uid"`
+			SessionID string `json:"session_id"`
 		}
 		if err := json.Unmarshal(body, &response); err != nil {
 			return nil, fmt.Errorf("event listener start response: %w", err)
 		}
-		if !response.Connected || response.Alias != s.environment.Alias || response.UID != s.uid {
+		s.sessionID = response.SessionID
+		if !response.Connected || response.Alias != s.environment.Alias || response.UID != s.uid || response.SessionID == "" {
 			return nil, errors.New("event listener start response identity or connection mismatch")
 		}
 		return nil, nil
 	}
-	var items []creator.ListenerDelivery
-	if err := json.Unmarshal(body, &items); err != nil {
-		return nil, fmt.Errorf("event listener poll response: %w", err)
-	}
-	if items == nil {
-		return nil, errors.New("event listener poll response must be an array")
-	}
-	for _, d := range items {
-		if err := d.Validate(); err != nil {
-			return nil, fmt.Errorf("event listener invalid %s delivery: %w", d.Kind, err)
-		}
-	}
-	return items, nil
+	return nil, creator.ErrInvalid
 }
-func (s gatewayListenerSession) Start(ctx context.Context) ([]creator.ListenerDelivery, error) {
-	return s.request(ctx, "start", nil)
-}
-func (s gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creator.ListenerDelivery, error) {
-	if s.pollInterval > 0 && !waitListener(ctx, s.pollInterval) {
-		return nil, ctx.Err()
-	}
-	items, err := s.request(ctx, "poll", ack)
-	if err != nil {
+func (s *gatewayListenerSession) Start(ctx context.Context) ([]creator.ListenerDelivery, error) {
+	if _, err := s.request(ctx, "start"); err != nil {
 		return nil, err
 	}
+	params := gatewayGenerationPayload(s.environment)
+	params["alias"], params["uid"] = s.environment.Alias, s.uid
+	var err error
+	params["listener_generation"] = s.generation
+	params["session_id"] = s.sessionID
+	subscribeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	s.subscription, err = acquireEventSubscription(subscribeCtx, s.target, params)
+	if err != nil {
+		cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+		defer cleanupCancel()
+		_, cleanupErr := s.request(cleanupCtx, "stop")
+		return nil, errors.Join(err, cleanupErr)
+	}
+	return nil, nil
+}
+func (s *gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creator.ListenerDelivery, error) {
+	return s.subscription.poll(ctx, ack)
+}
+func (s *gatewayListenerSession) enrich(ctx context.Context, items []creator.ListenerDelivery) ([]creator.ListenerDelivery, error) {
 	if s.coverFiles != nil {
 		browser := creatorGatewayBrowser{gateway: s.target, environment: s.environment}
 		fetch := func(ctx context.Context, target string) (string, []byte, error) {
@@ -149,8 +153,11 @@ func (s gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creat
 	}
 	return items, nil
 }
-func (s gatewayListenerSession) Stop(ctx context.Context) error {
-	_, err := s.request(ctx, "stop", nil)
+func (s *gatewayListenerSession) Stop(ctx context.Context) error {
+	if s.subscription != nil {
+		s.subscription.close()
+	}
+	_, err := s.request(ctx, "stop")
 	return err
 }
 
@@ -168,6 +175,9 @@ func runListenerSession(ctx context.Context, store listenerWriter, accountID, ge
 		}
 		err = errors.Join(err, stopErr)
 	}()
+	coverCtx, stopCovers := context.WithCancel(ctx)
+	defer stopCovers()
+	queueCovers := startListenerCoverWorker(coverCtx, store, accountID, generation, session)
 	healthStatus, healthReason := "starting", ""
 	for {
 		if err := ctx.Err(); err != nil {
@@ -189,6 +199,7 @@ func runListenerSession(ctx context.Context, store listenerWriter, accountID, ge
 					}
 				}
 			}
+			submitCovers := queueCovers(items)
 			persistenceFailed := false
 			for {
 				err := store.RecordListenerDeliveries(ctx, accountID, generation, items)
@@ -216,6 +227,18 @@ func runListenerSession(ctx context.Context, store listenerWriter, accountID, ge
 					return err
 				}
 			}
+			savedAt := time.Now().UTC()
+			for _, item := range items {
+				if item.Notice != nil {
+					fields := logrus.Fields{"account_id": accountID, "generation": generation, "delivery_id": item.DeliveryID, "event_key": item.Notice.EventKey, "realtime_signal": item.RealtimeSignal, "gateway_received_at": item.Notice.GatewayReceivedAt, "saved_at": savedAt.Format(time.RFC3339Nano)}
+					if item.Notice.GatewayReceivedAt != nil {
+						fields["gateway_to_save_clock_delta_ms"] = savedAt.Sub(*item.Notice.GatewayReceivedAt).Milliseconds()
+					}
+					logrus.WithFields(fields).Info("listener event committed")
+				}
+			}
+			publishEventChanges(ctx)
+			submitCovers()
 			logrus.WithFields(logrus.Fields{"account_id": accountID, "generation": generation, "delivery_count": len(items), "last_delivery_id": items[len(items)-1].DeliveryID}).Debug("account event deliveries committed")
 			for _, d := range items {
 				ack = append(ack, d.DeliveryID)
@@ -259,8 +282,8 @@ func listenerSessionForAccount(ctx context.Context, store listenerStore, account
 		return gatewayListenerSession{}, err
 	}
 	coverFiles, _ := store.(eventCoverFiles)
-	return gatewayListenerSession{target: target, environment: environment, uid: profile.PlatformAccountKey, coverFiles: coverFiles, resources: accounts,
-		boundaryAt: state.BoundaryAt, checkpoints: state.Checkpoints, pollInterval: 5 * time.Second}, nil
+	return gatewayListenerSession{target: target, environment: environment, uid: profile.PlatformAccountKey, generation: state.Generation, enabled: state.Enabled, coverFiles: coverFiles, resources: accounts,
+		boundaryAt: state.BoundaryAt, checkpoints: state.Checkpoints}, nil
 }
 
 func runAccountEventListener(ctx context.Context, store listenerStore, accounts HubStore, state creator.ListenerState) {
@@ -268,9 +291,13 @@ func runAccountEventListener(ctx context.Context, store listenerStore, accounts
 	update := func(status, reason string) {
 		writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
 		defer cancel()
-		if err := store.UpdateListenerStatus(writeCtx, state.AccountID, state.Generation, status, reason); err != nil && !errors.Is(err, creator.ErrConflict) {
+		err := store.UpdateListenerStatus(writeCtx, state.AccountID, state.Generation, status, reason)
+		if err != nil && !errors.Is(err, creator.ErrConflict) {
 			log.WithError(err).Error("persist event listener status failed")
 		}
+		if err == nil {
+			publishEventChanges(writeCtx)
+		}
 	}
 	for ctx.Err() == nil {
 		// Reload the last committed checkpoint after any failed request or restart.
@@ -313,7 +340,7 @@ func runAccountEventListener(ctx context.Context, store listenerStore, accounts
 			if err == nil {
 				if state.Enabled {
 					log.Info("account event listener starting")
-					err = runListenerSession(ctx, store, state.AccountID, state.Generation, session)
+					err = runListenerSession(ctx, store, state.AccountID, state.Generation, &session)
 				} else {
 					err = session.Stop(ctx)
 					if err == nil {
@@ -332,8 +359,8 @@ func runAccountEventListener(ctx context.Context, store listenerStore, accounts
 			return
 		}
 		update("error", err.Error())
-		log.WithError(err).Error("account event listener failed; retry in 10 seconds")
-		if !waitListener(ctx, 10*time.Second) {
+		log.WithError(err).Error("account event listener failed; retry in 2 seconds")
+		if !waitListener(ctx, 2*time.Second) {
 			return
 		}
 	}

internal/controlplane/api/event_push_commit_test.go

@@ -0,0 +1,55 @@
+package api
+
+import (
+	"context"
+	"git.ipao.vip/rogee/creator-hub/internal/creator"
+	"testing"
+	"time"
+)
+
+type gatedListenerWriter struct {
+	listenerWriterStub
+	entered, release chan struct{}
+}
+
+func (s *gatedListenerWriter) RecordListenerDeliveries(ctx context.Context, id, generation string, items []creator.ListenerDelivery) error {
+	close(s.entered)
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-s.release:
+	}
+	return s.listenerWriterStub.RecordListenerDeliveries(ctx, id, generation, items)
+}
+func TestListenerPushIsPublishedOnlyAfterSuccessfulCommit(t *testing.T) {
+	updates, remove := creatorUpdates.subscribe()
+	defer remove()
+	writer := &gatedListenerWriter{entered: make(chan struct{}), release: make(chan struct{})}
+	session := &listenerSessionStub{initial: []creator.ListenerDelivery{{Kind: "open", DeliveryID: "saved"}}, pollErr: creator.ErrConflict}
+	result := make(chan error, 1)
+	go func() { result <- runListenerSession(context.Background(), writer, "account", "generation", session) }()
+	select {
+	case <-writer.entered:
+	case <-time.After(time.Second):
+		t.Fatal("persistence not started")
+	}
+	select {
+	case <-updates:
+		t.Fatal("published before commit")
+	default:
+	}
+	close(writer.release)
+	select {
+	case <-result:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not finish")
+	}
+	select {
+	case <-updates:
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("successful commit did not notify page")
+	}
+	if len(session.ack) != 1 || session.ack[0] != "saved" {
+		t.Fatal("saved delivery not acknowledged")
+	}
+}

internal/controlplane/api/event_updates.go

@@ -0,0 +1,6 @@
+package api
+
+import "context"
+
+// Reuse the existing creator SSE feed; the database remains the source of truth.
+func publishEventChanges(context.Context) { creatorUpdates.publish() }