inventory-plus-plus/server/sse/publisher.go

package sse

import (
	"context"
	"errors"
	"fmt"
	"path"
	"strings"
	"sync"
	"time"

	"github.com/gin-gonic/gin"

	"ruben/inventory2/logging"
)

type (
	UpdateNotificationPublisher struct {
		log             *logging.Logger
		queue           sender
		getAccountID    func(*gin.Context) int64
		trimBasePath    string
		basePathPattern string
	}

	// sender is satisfied by *Queue
	sender interface {
		Send(ctx context.Context, e Event) error
	}
)

func (q *Queue) NewUpdateNotificationPublisher(
	log *logging.Logger,
	getAccountID func(*gin.Context) int64,
) *UpdateNotificationPublisher {
	return &UpdateNotificationPublisher{
		log:          log,
		queue:        q,
		getAccountID: getAccountID,
	}
}

// Trim produces an *UpdateNotificationPublisher with the pathPattern
// trimmed from events otherwise published by p.
func (p *UpdateNotificationPublisher) Trim(pathPattern string) *UpdateNotificationPublisher {
	p2 := *p
	p2.trimBasePath = path.Join(p2.trimBasePath, pathPattern)
	return &p2
}

// Group produces an *UpdateNotificationPublisher with the pathPattern
// appended to the base path used by p, if any, in publishing events.
// If no base path was set to p, then pathPattern becomes the base path.
func (p *UpdateNotificationPublisher) Group(pathPattern string) *UpdateNotificationPublisher {
	p2 := *p
	p2.basePathPattern = path.Join(p2.basePathPattern, pathPattern)
	return &p2
}

// Publish constructs a gin middleware.
// It must used with and after middleware puts in the Identity into the gin.Context.
func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFunc {
	trimBasePathSegs := getPathSegments(p.trimBasePath)
	trimmedBasePathPattern := path.Join(p.basePathPattern, pathPattern)
	trimmedBasePathPatternSegs := getPathSegments(trimmedBasePathPattern)
	fullBasePathPatternSegs := append(trimBasePathSegs, trimmedBasePathPatternSegs...)

	return func(c *gin.Context) {
		reqPathSegs := getPathSegments(c.Request.URL.Path)

		c.Next()
		if len(c.Errors) > 0 {
			return
		}
		if len(reqPathSegs) < len(fullBasePathPatternSegs) {
			p.log.Errorf("req path is less that the full path: %v, %v", c.Request.URL.Path, path.Join(p.trimBasePath, p.basePathPattern, pathPattern))
			return
		}

		// if the path is a subpath, then emit events along the subpath.

		for i, s := range trimmedBasePathPatternSegs {
			if isWildcard := s[0] == ':'; isWildcard {
				continue
			}

			rs := reqPathSegs[i]
			if isSubpath := s != rs; isSubpath {
				continue
			}

			return
		}

		topEventParts := reqPathSegs[len(trimBasePathSegs):len(fullBasePathPatternSegs)]
		topEvent := strings.Join(topEventParts, "_")

		events := make([]string, len(reqPathSegs)-len(fullBasePathPatternSegs)+1)
		events[0] = topEvent
		parentEvent := topEvent
		for i, s := range reqPathSegs[len(fullBasePathPatternSegs):] {
			e := parentEvent + "_" + s
			events[i+1] = e
			parentEvent = e
		}

		acctID := p.getAccountID(c)
		for _, e := range events {
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
				defer cancel()

				if err := p.queue.Send(ctx, Event{
					AccountID: acctID,
					Type:      e,
					Data:      []byte(fmt.Sprintf(`{"eventType": %q}`, e)),
				}); err != nil {
					p.log.Errorf("failed to send sse event to listener: %v", err)
				}
			}()
		}
	}
}

func (p *UpdateNotificationPublisher) Push(ctx context.Context, acctID int64, eventTypes ...string) error {
	var wg sync.WaitGroup
	wg.Add(len(eventTypes))

	errs := make([]error, len(eventTypes))
	for i, e := range eventTypes {
		i := i
		e := e

		go func() {
			defer wg.Done()
			if err := p.queue.Send(ctx, Event{
				AccountID: acctID,
				Type:      e,
				Data:      []byte(fmt.Sprintf(`{"eventType": %q}`, e)),
			}); err != nil {
				errs[i] = fmt.Errorf("failed to send sse event to listener: %w", err)
			}
		}()
	}

	wg.Wait()

	return errors.Join(errs...)

	return nil
}

func getPathSegments(p string) []string {
	p = path.Clean(p)
	if p == "" || p == "." || p == "/" {
		return nil
	}

	return strings.Split(strings.Trim(p, "/"), "/")
}