kaiyan-api: implement api server

cmd/kaiyan-api/main.go

@@ -0,0 +1,167 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"time"
+
+	"github.com/go-json-experiment/json"
+	"zombiezen.com/go/sqlite"
+	"zombiezen.com/go/sqlite/sqlitex"
+
+	"git.sunturtle.xyz/zephyr/kaiyan/emote"
+	"git.sunturtle.xyz/zephyr/kaiyan/emote/sqlitestore"
+)
+
+func main() {
+	var (
+		listen string
+		dburi  string
+	)
+	flag.StringVar(&listen, "listen", ":80", "address to bind")
+	flag.StringVar(&dburi, "db", "", "SQLite database URI")
+	flag.Parse()
+	if dburi == "" {
+		fail("-db is mandatory")
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	go func() {
+		<-ctx.Done()
+		stop()
+	}()
+
+	opts := sqlitex.PoolOptions{
+		Flags: sqlite.OpenReadWrite | sqlite.OpenURI,
+	}
+	db, err := sqlitex.NewPool(dburi, opts)
+	if err != nil {
+		fail("%v", err)
+	}
+	st, err := sqlitestore.Open(ctx, db)
+	if err != nil {
+		fail("%v", err)
+	}
+
+	mux := http.NewServeMux()
+	mux.Handle("GET /twitch/{channel}", metrics(st))
+	// TODO(branden): GET /metrics
+	l, err := net.Listen("tcp", listen)
+	if err != nil {
+		fail(err.Error())
+	}
+	srv := http.Server{
+		Addr:              listen,
+		Handler:           mux,
+		ReadHeaderTimeout: 3 * time.Second,
+		WriteTimeout:      30 * time.Second,
+		IdleTimeout:       1 * time.Minute,
+		BaseContext:       func(l net.Listener) context.Context { return ctx },
+	}
+	go func() {
+		slog.InfoContext(ctx, "HTTP API", slog.Any("addr", l.Addr()))
+		// TODO(branden): tls? or do we just use caddy/nginx after all?
+		err := srv.Serve(l)
+		if err == http.ErrServerClosed {
+			return
+		}
+		slog.ErrorContext(ctx, "HTTP API closed", slog.Any("err", err))
+	}()
+	<-ctx.Done()
+	stop()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	if err := srv.Shutdown(ctx); err != nil {
+		slog.ErrorContext(ctx, "HTTP API shutdown", slog.Any("err", err))
+	}
+}
+
+func metrics(st *sqlitestore.Store) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		log := slog.With(slog.Group("request",
+			slog.String("host", r.Host),
+			slog.String("remote", r.RemoteAddr),
+		))
+		channel := r.PathValue("channel")
+		var start, end time.Time
+		if s := r.FormValue("start"); s != "" {
+			t, err := time.Parse(time.RFC3339, s)
+			if err != nil {
+				writerror(w, http.StatusBadRequest, err.Error())
+				return
+			}
+			start = t
+		} else {
+			writerror(w, http.StatusBadRequest, "start time is required")
+			return
+		}
+		if s := r.FormValue("end"); s != "" {
+			t, err := time.Parse(time.RFC3339, s)
+			if err != nil {
+				writerror(w, http.StatusBadRequest, err.Error())
+				return
+			}
+			end = t
+		} else {
+			// We can fill in with the current time.
+			end = time.Now().Truncate(time.Second)
+		}
+		if end.Before(start) {
+			// Be liberal and just swap them.
+			start, end = end, start
+		}
+		if end.Sub(start) > 366*24*time.Hour {
+			writerror(w, http.StatusBadRequest, "timespan is too long")
+			return
+		}
+		log.LogAttrs(r.Context(), slog.LevelInfo, "metrics",
+			slog.String("channel", channel),
+			slog.Time("start", start),
+			slog.Time("end", end),
+		)
+		m, err := st.Metrics(r.Context(), channel, start, end, nil)
+		if err != nil {
+			log.ErrorContext(r.Context(), "metrics", slog.Any("err", err))
+			writerror(w, http.StatusInternalServerError, "data unavailable")
+			return
+		}
+		log.LogAttrs(r.Context(), slog.LevelInfo, "done",
+			slog.String("channel", channel),
+			slog.Time("start", start),
+			slog.Time("end", end),
+		)
+		writedata(w, m)
+	}
+}
+
+func writerror(w http.ResponseWriter, status int, err string) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	v := struct {
+		Error string `json:"error"`
+	}{err}
+	json.MarshalWrite(w, &v)
+}
+
+func writedata(w http.ResponseWriter, data []emote.Metric) {
+	w.Header().Set("Content-Type", "application/json")
+	if len(data) == 0 {
+		w.Write([]byte(`{"data":[]}`))
+		return
+	}
+	v := struct {
+		Data []emote.Metric `json:"data"`
+	}{data}
+	json.MarshalWrite(w, &v)
+}
+
+func fail(format string, args ...any) {
+	fmt.Fprintf(os.Stderr, format+"\n", args...)
+	os.Exit(1)
+}

cmd/kaiyan-index/main.go

@@ -0,0 +1,182 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+	"zombiezen.com/go/sqlite"
+	"zombiezen.com/go/sqlite/sqlitex"
+
+	"git.sunturtle.xyz/zephyr/kaiyan/emote"
+	"git.sunturtle.xyz/zephyr/kaiyan/emote/sqlitestore"
+	"git.sunturtle.xyz/zephyr/kaiyan/queue"
+)
+
+func main() {
+	var (
+		dburi   string
+		streams []string
+	)
+	flag.StringVar(&dburi, "db", "", "SQLite database URI")
+	flag.Func("b", "Twitch broadcaster user ID", func(s string) error {
+		streams = append(streams, s)
+		_, err := strconv.Atoi(s)
+		return err
+	})
+	flag.Parse()
+	if flag.NArg() == 0 {
+		fail("provide broker addresses as args")
+	}
+	if dburi == "" {
+		fail("-db is mandatory")
+	}
+	if len(streams) == 0 {
+		fail("no broadcasters specified")
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	go func() {
+		<-ctx.Done()
+		stop()
+	}()
+
+	parsers := make(map[string]emote.Parser)
+	{
+		var mu sync.Mutex
+		var wg sync.WaitGroup
+		wg.Add(len(streams))
+		for _, b := range streams {
+			go func() {
+				defer wg.Done()
+				req, err := http.NewRequestWithContext(ctx, "GET", "https://7tv.io/v3/users/twitch/"+b, nil)
+				if err != nil {
+					fail("%v", err)
+				}
+				resp, err := http.DefaultClient.Do(req)
+				if err != nil {
+					fail("%v", err)
+				}
+				defer resp.Body.Close()
+				if resp.StatusCode != 200 {
+					slog.ErrorContext(ctx, "7tv", slog.String("broadcaster", b), slog.String("status", resp.Status))
+					return
+				}
+				em, err := emote.SevenTVv3(resp.Body)
+				if err != nil {
+					slog.ErrorContext(ctx, "7tv", slog.String("broadcaster", b), slog.Any("err", err))
+					return
+				}
+				mu.Lock()
+				parsers[b] = emote.NewParser(em...)
+				mu.Unlock()
+			}()
+		}
+		wg.Wait()
+	}
+
+	opts := sqlitex.PoolOptions{
+		Flags: sqlite.OpenCreate | sqlite.OpenReadWrite | sqlite.OpenURI,
+		PrepareConn: func(conn *sqlite.Conn) error {
+			pragmas := []string{
+				`PRAGMA journal_mode = WAL`,
+				`PRAGMA synchronous = NORMAL`,
+			}
+			for _, p := range pragmas {
+				s, _, err := conn.PrepareTransient(p)
+				if err != nil {
+					// If this function returns an error, then the pool will
+					// continue to invoke it for every connection.
+					// Explode violently instead.
+					panic(fmt.Errorf("couldn't set %s: %w", p, err))
+				}
+				if _, err := s.Step(); err != nil {
+					// This one is probably ok to retry, though.
+					return fmt.Errorf("couldn't run %s: %w", p, err)
+				}
+				if err := s.Finalize(); err != nil {
+					panic(fmt.Errorf("couldn't finalize statement for %s: %w", p, err))
+				}
+			}
+			return nil
+		},
+	}
+	db, err := sqlitex.NewPool(dburi, opts)
+	if err != nil {
+		fail("%v", err)
+	}
+	st, err := sqlitestore.Open(ctx, db)
+	if err != nil {
+		fail("%v", err)
+	}
+	q, err := kgo.NewClient(
+		kgo.SeedBrokers(flag.Args()...),
+		kgo.ConsumerGroup("kaiyan"),
+		kgo.ConsumeTopics(queue.Topic),
+		kgo.FetchIsolationLevel(kgo.ReadCommitted()),
+		kgo.GreedyAutoCommit(),
+	)
+	if err != nil {
+		fail("%v", err)
+	}
+
+	var msgs []queue.Message
+	var ems []emote.Emote
+	// TODO(branden): emote lists
+	for ctx.Err() == nil {
+		msgs, ems, err = batch(ctx, parsers, q, st, msgs[:0], ems[:0])
+		if err != nil {
+			fail("%v", err)
+		}
+	}
+	st.Close()
+}
+
+func batch(ctx context.Context, ps map[string]emote.Parser, q *kgo.Client, st *sqlitestore.Store, msgs []queue.Message, ems []emote.Emote) ([]queue.Message, []emote.Emote, error) {
+	slog.InfoContext(ctx, "receive")
+	msgs, err := queue.Recv(ctx, q, msgs)
+	switch {
+	case err == nil: // do nothing
+	case ctx.Err() != nil:
+		// Exiting normally.
+		// Return no error and let the caller check the context.
+		return nil, nil, nil
+	default:
+		// TODO(branden): there are a variety of errors that are non-fatal;
+		// we should check for them and continue instead of giving up
+		return nil, nil, fmt.Errorf("queue recv: %w", err)
+	}
+	n := 0
+	slog.InfoContext(ctx, "process", slog.Int("count", len(msgs)))
+	for _, msg := range msgs {
+		p := ps[msg.Channel]
+		text := msg.Text
+		for {
+			em, rest := p.Next(text)
+			if em.ID == "" {
+				break
+			}
+			ems = append(ems, em)
+			text = rest
+		}
+		if err := st.Record(ctx, msg.Channel, msg.ID, msg.Sender.String(), time.Unix(0, msg.Timestamp), ems); err != nil {
+			return nil, nil, err
+		}
+		n += len(ems)
+	}
+	slog.InfoContext(ctx, "recorded", slog.Int("emotes", n))
+	return msgs, ems, nil
+}
+
+func fail(format string, args ...any) {
+	fmt.Fprintf(os.Stderr, format+"\n", args...)
+	os.Exit(1)
+}

cmd/kaiyan-ingest/main.go

@@ -125,11 +125,21 @@ func webhook(q *kgo.Client, secret []byte, errs chan<- error) http.HandlerFunc {
 				slog.String("broadcaster", ev.Event.Broadcaster),
 				slog.String("id", ev.Event.ID),
 			)
+			// Use the external clock for timestamps.
+			tm, err := time.Parse(time.RFC3339Nano, r.Header.Get("Twitch-Eventsub-Message-Timestamp"))
+			if err != nil {
+				log.LogAttrs(ctx, slog.LevelWarn, "bad timestamp",
+					slog.String("timestamp", r.Header.Get("Twitch-Eventsub-Message-Timestamp")),
+					slog.Any("err", err),
+				)
+				tm = time.Now()
+			}
 			msg := queue.Message{
-				ID:      ev.Event.ID,
-				Channel: ev.Event.Broadcaster,
-				Sender:  queue.Sender(secret, ev.Event.Broadcaster, ev.Event.Chatter),
-				Text:    ev.Event.Message.Text,
+				ID:        ev.Event.ID,
+				Channel:   ev.Event.Broadcaster,
+				Sender:    queue.Sender(secret, ev.Event.Broadcaster, ev.Event.Chatter),
+				Timestamp: tm.UnixNano(),
+				Text:      ev.Event.Message.Text,
 			}
 			// TODO(branden): the context in the http request cancels once this
 			// handler returns, which means the producer doesn't relay it.

cmd/reqecho/main.go

@@ -18,6 +18,6 @@ func main() {
 		if _, err := f.Write(b); err != nil {
 			panic(err)
 		}
-		os.Exit(0)
+		w.WriteHeader(http.StatusNoContent)
 	}))
 }

emote/7tv.go

@@ -0,0 +1,44 @@
+package emote
+
+import (
+	"io"
+
+	"github.com/go-json-experiment/json"
+)
+
+// seventvUser has the fields of a 7TV user model which are relevant to Kaiyan.
+type seventvUser struct {
+	EmoteSet emoteSet `json:"emote_set"`
+}
+
+type emoteSet struct {
+	Emotes []seventvEmote `json:"emotes"`
+}
+
+type seventvEmote struct {
+	ID   string `json:"id"`
+	Name string `json:"name"`
+	// 7TV provides the CDN URL as well, but it's nested in further objects,
+	// not a complete URL, and it's also trivially derivable from the ID.
+}
+
+// SevenTVv3 parses the emotes from a 7TV v3/users/{platform}/{platform_id}
+// response body.
+func SevenTVv3(r io.Reader) ([]Emote, error) {
+	var u seventvUser
+	if err := json.UnmarshalRead(r, &u); err != nil {
+		return nil, err
+	}
+	ems := make([]Emote, 0, len(u.EmoteSet.Emotes))
+	for _, em := range u.EmoteSet.Emotes {
+		v := Emote{
+			ID:     em.ID,
+			Name:   em.Name,
+			Source: "7TV",
+			Link:   "https://7tv.app/emotes/" + em.ID,
+			Image:  "https://cdn.7tv.app/emote/" + em.ID + "/4x.webp",
+		}
+		ems = append(ems, v)
+	}
+	return ems, nil
+}

emote/7tv_test.go

@@ -0,0 +1,46 @@
+package emote_test
+
+import (
+	_ "embed"
+	"strings"
+	"testing"
+
+	"git.sunturtle.xyz/zephyr/kaiyan/emote"
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestSevenTVv3(t *testing.T) {
+	got, err := emote.SevenTVv3(strings.NewReader(seventvTwin))
+	if err != nil {
+		t.Error(err)
+	}
+	want := []emote.Emote{
+		{
+			ID:     "01HBHC9CE00007KVA84SNERRE8",
+			Name:   "powerSip",
+			Source: "7TV",
+			Link:   "https://7tv.app/emotes/01HBHC9CE00007KVA84SNERRE8",
+			Image:  "https://cdn.7tv.app/emote/01HBHC9CE00007KVA84SNERRE8/4x.webp",
+		},
+		{
+			ID:     "01JDTNVJADW1R4YZB3F85VVT9T",
+			Name:   "Jin",
+			Source: "7TV",
+			Link:   "https://7tv.app/emotes/01JDTNVJADW1R4YZB3F85VVT9T",
+			Image:  "https://cdn.7tv.app/emote/01JDTNVJADW1R4YZB3F85VVT9T/4x.webp",
+		},
+		{
+			ID:     "01JE08SKFXM94FTG52GVMGFQZW",
+			Name:   "IMissTwin",
+			Source: "7TV",
+			Link:   "https://7tv.app/emotes/01JE08SKFXM94FTG52GVMGFQZW",
+			Image:  "https://cdn.7tv.app/emote/01JE08SKFXM94FTG52GVMGFQZW/4x.webp",
+		},
+	}
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("wrong emotes (+got/-want):\n%s", diff)
+	}
+}
+
+//go:embed testdata/twoinone1_.7tvv3.json
+var seventvTwin string

emote/emote.go

@@ -8,15 +8,15 @@ import (
 // Emote is the information Kaiyan needs about an emote.
 type Emote struct {
 	// ID is the emote ID per the source.
-	ID string
+	ID string `json:"id"`
 	// Name is the text of the emote as would be parsed from message text.
-	Name string
+	Name string `json:"name"`
 	// Source is the name of the emote source, e.g. "7TV", "Twitch:cirno_tv", &c.
-	Source string
+	Source string `json:"source"`
 	// Link is a hyperlink to manage the emote.
-	Link string
+	Link string `json:"link"`
 	// Image is a hyperlink to the emote image of any size.
-	Image string
+	Image string `json:"image"`
 }
 
 // Parser finds emotes in a message.

emote/store.go

@@ -16,11 +16,11 @@ type Store interface {
 // Metric is the metrics for a single emote.
 type Metric struct {
 	// Emote is the emote that the metrics describe.
-	Emote Emote
+	Emote Emote `json:"emote"`
 	// Tokens is the total number of occurrences of the emote.
-	Tokens int64
+	Tokens int64 `json:"tokens"`
 	// Messages is the number of unique messages which contained the emote.
-	Messages int64
+	Messages int64 `json:"messages"`
 	// Users is the number of users who used the emote.
-	Users int64
+	Users int64 `json:"users"`
 }

emote/testdata/twoinone1_.7tvv3.json

@@ -0,0 +1,538 @@
+{
+	"id": "116723523",
+	"platform": "TWITCH",
+	"username": "twoinone1_",
+	"display_name": "twoinone1_",
+	"linked_at": 1730685664536,
+	"emote_capacity": 1000,
+	"emote_set_id": "01JD8YZCMPF628X8DZ7NAYFG8C",
+	"emote_set": {
+		"id": "01JD8YZCMPF628X8DZ7NAYFG8C",
+		"name": "twoinone1_'s Emotes",
+		"flags": 0,
+		"tags": [],
+		"immutable": false,
+		"privileged": false,
+		"emotes": [
+			{
+				"id": "01HBHC9CE00007KVA84SNERRE8",
+				"name": "powerSip",
+				"flags": 0,
+				"timestamp": 1696021656000,
+				"actor_id": "01JBTEEZ8R6A1J8RMZ581A1V02",
+				"data": {
+					"id": "01HBHC9CE00007KVA84SNERRE8",
+					"name": "powerSip",
+					"flags": 0,
+					"tags": [
+						"csm",
+						"weeb",
+						"anime",
+						"sip"
+					],
+					"lifecycle": 3,
+					"state": [
+						"LISTED"
+					],
+					"listed": true,
+					"animated": false,
+					"owner": {
+						"id": "01FBXP96C0000APPS8GMQY5HFA",
+						"username": "kodyparbs",
+						"display_name": "Kodyparbs",
+						"avatar_url": "https://static-cdn.jtvnw.net/jtv_user_pictures/88453ec4-d53f-4413-a1c0-57319ccc679a-profile_image-300x300.png",
+						"style": {},
+						"role_ids": [
+							"01G68MMQFR0007J6GNM9E2M0TM"
+						],
+						"connections": [
+							{
+								"id": "417084954",
+								"platform": "TWITCH",
+								"username": "kodyparbs",
+								"display_name": "Kodyparbs",
+								"linked_at": 1627715312000,
+								"emote_capacity": 1000,
+								"emote_set_id": "01FBXP96C0000APPS8GMQY5HFA"
+							}
+						]
+					},
+					"host": {
+						"url": "//cdn.7tv.app/emote/01HBHC9CE00007KVA84SNERRE8",
+						"files": [
+							{
+								"name": "1x.webp",
+								"static_name": "1x.webp",
+								"width": 32,
+								"height": 32,
+								"frame_count": 1,
+								"size": 2568,
+								"format": "WEBP"
+							},
+							{
+								"name": "2x.webp",
+								"static_name": "2x.webp",
+								"width": 64,
+								"height": 64,
+								"frame_count": 1,
+								"size": 6388,
+								"format": "WEBP"
+							},
+							{
+								"name": "3x.webp",
+								"static_name": "3x.webp",
+								"width": 96,
+								"height": 96,
+								"frame_count": 1,
+								"size": 10816,
+								"format": "WEBP"
+							},
+							{
+								"name": "4x.webp",
+								"static_name": "4x.webp",
+								"width": 128,
+								"height": 128,
+								"frame_count": 1,
+								"size": 16062,
+								"format": "WEBP"
+							},
+							{
+								"name": "1x.avif",
+								"static_name": "1x.avif",
+								"width": 32,
+								"height": 32,
+								"frame_count": 1,
+								"size": 1632,
+								"format": "AVIF"
+							},
+							{
+								"name": "2x.avif",
+								"static_name": "2x.avif",
+								"width": 64,
+								"height": 64,
+								"frame_count": 1,
+								"size": 3607,
+								"format": "AVIF"
+							},
+							{
+								"name": "3x.avif",
+								"static_name": "3x.avif",
+								"width": 96,
+								"height": 96,
+								"frame_count": 1,
+								"size": 5813,
+								"format": "AVIF"
+							},
+							{
+								"name": "4x.avif",
+								"static_name": "4x.avif",
+								"width": 128,
+								"height": 128,
+								"frame_count": 1,
+								"size": 8011,
+								"format": "AVIF"
+							},
+							{
+								"name": "1x.png",
+								"static_name": "1x.png",
+								"width": 32,
+								"height": 32,
+								"frame_count": 1,
+								"size": 2919,
+								"format": "PNG"
+							},
+							{
+								"name": "2x.png",
+								"static_name": "2x.png",
+								"width": 64,
+								"height": 64,
+								"frame_count": 1,
+								"size": 8672,
+								"format": "PNG"
+							},
+							{
+								"name": "3x.png",
+								"static_name": "3x.png",
+								"width": 96,
+								"height": 96,
+								"frame_count": 1,
+								"size": 15712,
+								"format": "PNG"
+							},
+							{
+								"name": "4x.png",
+								"static_name": "4x.png",
+								"width": 128,
+								"height": 128,
+								"frame_count": 1,
+								"size": 23966,
+								"format": "PNG"
+							}
+						]
+					}
+				},
+				"origin_id": null
+			},
+			{
+				"id": "01JDTNVJADW1R4YZB3F85VVT9T",
+				"name": "Jin",
+				"flags": 0,
+				"timestamp": 1732840900941,
+				"actor_id": "01JBTEEZ8R6A1J8RMZ581A1V02",
+				"data": {
+					"id": "01JDTNVJADW1R4YZB3F85VVT9T",
+					"name": "Jin",
+					"flags": 0,
+					"lifecycle": 3,
+					"state": [
+						"LISTED"
+					],
+					"listed": true,
+					"animated": false,
+					"owner": {
+						"id": "01JBTEEZ8R6A1J8RMZ581A1V02",
+						"username": "twoinone1_",
+						"display_name": "twoinone1_",
+						"avatar_url": "https://static-cdn.jtvnw.net/jtv_user_pictures/3d606741-9960-484d-a484-1aa9a0f002b6-profile_image-300x300.png",
+						"style": {
+							"color": -5635841,
+							"paint_id": "01JQM68WYPTC7PFSE1QZKGWSTX"
+						},
+						"role_ids": [
+							"01F37R3RFR0000K96678WEQT01",
+							"01G68MMQFR0007J6GNM9E2M0TM"
+						],
+						"connections": [
+							{
+								"id": "116723523",
+								"platform": "TWITCH",
+								"username": "twoinone1_",
+								"display_name": "twoinone1_",
+								"linked_at": 1730685664536,
+								"emote_capacity": 1000,
+								"emote_set_id": "01JD8YZCMPF628X8DZ7NAYFG8C"
+							}
+						]
+					},
+					"host": {
+						"url": "//cdn.7tv.app/emote/01JDTNVJADW1R4YZB3F85VVT9T",
+						"files": [
+							{
+								"name": "1x.webp",
+								"static_name": "1x.webp",
+								"width": 74,
+								"height": 32,
+								"frame_count": 1,
+								"size": 774,
+								"format": "WEBP"
+							},
+							{
+								"name": "2x.webp",
+								"static_name": "2x.webp",
+								"width": 148,
+								"height": 64,
+								"frame_count": 1,
+								"size": 1768,
+								"format": "WEBP"
+							},
+							{
+								"name": "3x.webp",
+								"static_name": "3x.webp",
+								"width": 222,
+								"height": 96,
+								"frame_count": 1,
+								"size": 2880,
+								"format": "WEBP"
+							},
+							{
+								"name": "4x.webp",
+								"static_name": "4x.webp",
+								"width": 296,
+								"height": 128,
+								"frame_count": 1,
+								"size": 3916,
+								"format": "WEBP"
+							},
+							{
+								"name": "1x.avif",
+								"static_name": "1x.avif",
+								"width": 74,
+								"height": 32,
+								"frame_count": 1,
+								"size": 1192,
+								"format": "AVIF"
+							},
+							{
+								"name": "2x.avif",
+								"static_name": "2x.avif",
+								"width": 148,
+								"height": 64,
+								"frame_count": 1,
+								"size": 2598,
+								"format": "AVIF"
+							},
+							{
+								"name": "3x.avif",
+								"static_name": "3x.avif",
+								"width": 222,
+								"height": 96,
+								"frame_count": 1,
+								"size": 3963,
+								"format": "AVIF"
+							},
+							{
+								"name": "4x.avif",
+								"static_name": "4x.avif",
+								"width": 296,
+								"height": 128,
+								"frame_count": 1,
+								"size": 5406,
+								"format": "AVIF"
+							},
+							{
+								"name": "1x.png",
+								"static_name": "1x.png",
+								"width": 74,
+								"height": 32,
+								"frame_count": 1,
+								"size": 4913,
+								"format": "PNG"
+							},
+							{
+								"name": "2x.png",
+								"static_name": "2x.png",
+								"width": 148,
+								"height": 64,
+								"frame_count": 1,
+								"size": 13376,
+								"format": "PNG"
+							},
+							{
+								"name": "3x.png",
+								"static_name": "3x.png",
+								"width": 222,
+								"height": 96,
+								"frame_count": 1,
+								"size": 24123,
+								"format": "PNG"
+							},
+							{
+								"name": "4x.png",
+								"static_name": "4x.png",
+								"width": 296,
+								"height": 128,
+								"frame_count": 1,
+								"size": 34810,
+								"format": "PNG"
+							}
+						]
+					}
+				},
+				"origin_id": null
+			},
+			{
+				"id": "01JE08SKFXM94FTG52GVMGFQZW",
+				"name": "IMissTwin",
+				"flags": 0,
+				"timestamp": 1733028531709,
+				"actor_id": "01JBTEEZ8R6A1J8RMZ581A1V02",
+				"data": {
+					"id": "01JE08SKFXM94FTG52GVMGFQZW",
+					"name": "IMissTwin",
+					"flags": 0,
+					"lifecycle": 3,
+					"state": [
+						"LISTED"
+					],
+					"listed": true,
+					"animated": false,
+					"owner": {
+						"id": "01JBTEEZ8R6A1J8RMZ581A1V02",
+						"username": "twoinone1_",
+						"display_name": "twoinone1_",
+						"avatar_url": "https://static-cdn.jtvnw.net/jtv_user_pictures/3d606741-9960-484d-a484-1aa9a0f002b6-profile_image-300x300.png",
+						"style": {
+							"color": -5635841,
+							"paint_id": "01JQM68WYPTC7PFSE1QZKGWSTX"
+						},
+						"role_ids": [
+							"01F37R3RFR0000K96678WEQT01",
+							"01G68MMQFR0007J6GNM9E2M0TM"
+						],
+						"connections": [
+							{
+								"id": "116723523",
+								"platform": "TWITCH",
+								"username": "twoinone1_",
+								"display_name": "twoinone1_",
+								"linked_at": 1730685664536,
+								"emote_capacity": 1000,
+								"emote_set_id": "01JD8YZCMPF628X8DZ7NAYFG8C"
+							}
+						]
+					},
+					"host": {
+						"url": "//cdn.7tv.app/emote/01JE08SKFXM94FTG52GVMGFQZW",
+						"files": [
+							{
+								"name": "1x.webp",
+								"static_name": "1x.webp",
+								"width": 44,
+								"height": 32,
+								"frame_count": 1,
+								"size": 748,
+								"format": "WEBP"
+							},
+							{
+								"name": "2x.webp",
+								"static_name": "2x.webp",
+								"width": 88,
+								"height": 64,
+								"frame_count": 1,
+								"size": 1964,
+								"format": "WEBP"
+							},
+							{
+								"name": "3x.webp",
+								"static_name": "3x.webp",
+								"width": 132,
+								"height": 96,
+								"frame_count": 1,
+								"size": 3444,
+								"format": "WEBP"
+							},
+							{
+								"name": "4x.webp",
+								"static_name": "4x.webp",
+								"width": 176,
+								"height": 128,
+								"frame_count": 1,
+								"size": 5210,
+								"format": "WEBP"
+							},
+							{
+								"name": "1x.avif",
+								"static_name": "1x.avif",
+								"width": 44,
+								"height": 32,
+								"frame_count": 1,
+								"size": 1408,
+								"format": "AVIF"
+							},
+							{
+								"name": "2x.avif",
+								"static_name": "2x.avif",
+								"width": 88,
+								"height": 64,
+								"frame_count": 1,
+								"size": 3291,
+								"format": "AVIF"
+							},
+							{
+								"name": "3x.avif",
+								"static_name": "3x.avif",
+								"width": 132,
+								"height": 96,
+								"frame_count": 1,
+								"size": 5577,
+								"format": "AVIF"
+							},
+							{
+								"name": "4x.avif",
+								"static_name": "4x.avif",
+								"width": 176,
+								"height": 128,
+								"frame_count": 1,
+								"size": 8071,
+								"format": "AVIF"
+							},
+							{
+								"name": "1x.png",
+								"static_name": "1x.png",
+								"width": 44,
+								"height": 32,
+								"frame_count": 1,
+								"size": 4904,
+								"format": "PNG"
+							},
+							{
+								"name": "2x.png",
+								"static_name": "2x.png",
+								"width": 88,
+								"height": 64,
+								"frame_count": 1,
+								"size": 16973,
+								"format": "PNG"
+							},
+							{
+								"name": "3x.png",
+								"static_name": "3x.png",
+								"width": 132,
+								"height": 96,
+								"frame_count": 1,
+								"size": 34591,
+								"format": "PNG"
+							},
+							{
+								"name": "4x.png",
+								"static_name": "4x.png",
+								"width": 176,
+								"height": 128,
+								"frame_count": 1,
+								"size": 56643,
+								"format": "PNG"
+							}
+						]
+					}
+				},
+				"origin_id": null
+			}
+		],
+		"emote_count": 3,
+		"capacity": 1000,
+		"owner": null
+	},
+	"user": {
+		"id": "01JBTEEZ8R6A1J8RMZ581A1V02",
+		"username": "twoinone1_",
+		"display_name": "twoinone1_",
+		"created_at": 1730685664536,
+		"avatar_url": "https://static-cdn.jtvnw.net/jtv_user_pictures/3d606741-9960-484d-a484-1aa9a0f002b6-profile_image-300x300.png",
+		"style": {
+			"color": -5635841,
+			"paint_id": "01JQM68WYPTC7PFSE1QZKGWSTX"
+		},
+		"emote_sets": [
+			{
+				"id": "01JD8YZCMPF628X8DZ7NAYFG8C",
+				"name": "twoinone1_'s Emotes",
+				"flags": 0,
+				"tags": [],
+				"capacity": 1000
+			},
+			{
+				"id": "01JR4JFB07TJHTFCCF2BDZM8MH",
+				"name": "Personal Emote Set",
+				"flags": 4,
+				"tags": [],
+				"capacity": 5
+			}
+		],
+		"roles": [
+			"01G68MMQFR0007J6GNM9E2M0TM",
+			"01F37R3RFR0000K96678WEQT01"
+		],
+		"connections": [
+			{
+				"id": "116723523",
+				"platform": "TWITCH",
+				"username": "twoinone1_",
+				"display_name": "twoinone1_",
+				"linked_at": 1730685664536,
+				"emote_capacity": 1000,
+				"emote_set_id": "01JD8YZCMPF628X8DZ7NAYFG8C",
+				"emote_set": null
+			}
+		]
+	}
+}

queue/message.go

@@ -18,6 +18,9 @@ type Message struct {
 	Channel string `json:"ch"`
 	// Sender is an obfuscated identifier for the sending user.
 	Sender sender `json:"u"`
+	// Timestamp is the time at which the message was sent in nanoseconds since
+	// the Unix epoch.
+	Timestamp int64 `json:"ts"`
 	// Text is the message content.
 	Text string `json:"t"`
 }

queue/message_test.go

@@ -21,10 +21,11 @@ func (p *spyProducer) Produce(ctx context.Context, rec *kgo.Record, promise func
 
 func TestSend(t *testing.T) {
 	msg := queue.Message{
-		ID:      "bocchi",
-		Channel: "kessoku",
-		Sender:  queue.Sender(make([]byte, 16), "kessoku", "ryō"),
-		Text:    "bocchi the rock!",
+		ID:        "bocchi",
+		Channel:   "kessoku",
+		Sender:    queue.Sender(make([]byte, 16), "kessoku", "ryō"),
+		Timestamp: 1,
+		Text:      "bocchi the rock!",
 	}
 	var q spyProducer
 	errs := make(chan error, 1)