Merge pull request #16 from jacisjake/task/9-nest-message-log

(#9) Nest message log + replay

nestlog.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+)
+
+// Message is one entry in a nest's conversation, shaped by the EVENTS.md
+// contract. IDs are assigned by the store; MsgID is "<nest_id>-<n>" and is
+// what the contract exposes.
+type Message struct {
+	NestID    string `json:"nest_id"`
+	ID        int    `json:"id"`
+	Kind      string `json:"kind"`
+	Author    string `json:"author"`
+	Text      string `json:"text"`
+	To        string `json:"to,omitempty"`
+	RefersTo  string `json:"refers_to,omitempty"`
+	TaggedBy  string `json:"tagged_by,omitempty"`
+	Timestamp string `json:"ts"`
+}
+
+func (m Message) MsgID() string { return fmt.Sprintf("%s-%d", m.NestID, m.ID) }
+
+// MessageStore is the append-only log of every nest's conversation, persisted
+// as JSON Lines (one file per nest) so a restart can replay it. Follows the
+// atomic-write style of NestStore.save().
+type MessageStore struct {
+	dir string
+	mu  sync.Mutex
+	log map[string][]Message // nest_id -> messages in append order
+}
+
+func NewMessageStore(dir string) *MessageStore {
+	return &MessageStore{dir: dir, log: map[string][]Message{}}
+}
+
+// LoadMessageStore reads existing logs from dir. A missing dir is empty, not
+// an error: a fresh server has no conversations yet.
+func LoadMessageStore(dir string) (*MessageStore, error) {
+	s := NewMessageStore(dir)
+	entries, err := os.ReadDir(dir)
+	if err != nil && !os.IsNotExist(err) {
+		return nil, err
+	}
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
+			continue
+		}
+		nestID := strings.TrimSuffix(e.Name(), ".jsonl")
+		data, err := os.ReadFile(filepath.Join(dir, e.Name()))
+		if err != nil {
+			return nil, fmt.Errorf("message log %s is unreadable: %w", e.Name(), err)
+		}
+		var msgs []Message
+		for _, line := range strings.Split(string(data), "\n") {
+			if strings.TrimSpace(line) == "" {
+				continue
+			}
+			var m Message
+			if err := json.Unmarshal([]byte(line), &m); err != nil {
+				return nil, fmt.Errorf("message log %s has a bad line: %w", e.Name(), err)
+			}
+			msgs = append(msgs, m)
+		}
+		s.log[nestID] = msgs
+	}
+	return s, nil
+}
+
+// Append assigns the next id, records the message and saves its nest's log.
+func (s *MessageStore) Append(m Message) (Message, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	m.ID = len(s.log[m.NestID]) + 1
+	s.log[m.NestID] = append(s.log[m.NestID], m)
+	if err := s.save(m.NestID); err != nil {
+		log := s.log[m.NestID]
+		s.log[m.NestID] = log[:len(log)-1]
+		return Message{}, err
+	}
+	return m, nil
+}
+
+// After returns the messages in nest with id greater than lastID, in order.
+func (s *MessageStore) After(nestID string, lastID int) []Message {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []Message
+	for _, m := range s.log[nestID] {
+		if m.ID > lastID {
+			out = append(out, m)
+		}
+	}
+	return out
+}
+
+// All returns every message in a nest, oldest first.
+func (s *MessageStore) All(nestID string) []Message { return s.After(nestID, 0) }
+
+func (s *MessageStore) save(nestID string) error {
+	if err := os.MkdirAll(s.dir, 0o755); err != nil {
+		return err
+	}
+	var sb strings.Builder
+	for _, m := range s.log[nestID] {
+		line, err := json.Marshal(m)
+		if err != nil {
+			return err
+		}
+		sb.Write(line)
+		sb.WriteByte('\n')
+	}
+	path := filepath.Join(s.dir, nestID+".jsonl")
+	tmp := path + ".tmp"
+	if err := os.WriteFile(tmp, []byte(sb.String()), 0o644); err != nil {
+		return err
+	}
+	return os.Rename(tmp, path)
+}

nestlog_test.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"path/filepath"
+	"testing"
+)
+
+func TestMessageLogAppendOrder(t *testing.T) {
+	s := NewMessageStore(filepath.Join(t.TempDir(), "messages"))
+	for _, m := range []Message{{NestID: "lab", Kind: "say", Author: "owner", Text: "a"}, {NestID: "lab", Kind: "request", Author: "owner", Text: "b"}, {NestID: "lab", Kind: "claim", Author: "Hugin", Text: "c"}} {
+		if _, err := s.Append(m); err != nil {
+			t.Fatal(err)
+		}
+	}
+	got := s.After("lab", 0)
+	if len(got) != 3 || got[0].ID != 1 || got[1].ID != 2 || got[2].ID != 3 || got[2].Kind != "claim" {
+		t.Fatalf("want ids 1,2,3 in order, got %+v", got)
+	}
+}
+
+func TestMessageLogReplayFrom(t *testing.T) {
+	s := NewMessageStore(filepath.Join(t.TempDir(), "messages"))
+	for i := 0; i < 5; i++ {
+		if _, err := s.Append(Message{NestID: "lab", Kind: "say", Author: "owner"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	got := s.After("lab", 2)
+	if len(got) != 3 || got[0].ID != 3 || got[2].ID != 5 {
+		t.Fatalf("want ids 3,4,5, got %+v", got)
+	}
+	if n := len(s.After("lab", 5)); n != 0 {
+		t.Fatalf("want nothing after the last id, got %d", n)
+	}
+}
+
+func TestMessageLogPersists(t *testing.T) {
+	dir := filepath.Join(t.TempDir(), "messages")
+	s := NewMessageStore(dir)
+	for _, m := range []Message{{NestID: "lab", Kind: "request", Author: "owner", Text: "do it"}, {NestID: "lab", Kind: "claim", Author: "Munin", Text: "mine", To: "owner", RefersTo: "lab-1"}} {
+		if _, err := s.Append(m); err != nil {
+			t.Fatal(err)
+		}
+	}
+	opened, err := LoadMessageStore(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := opened.After("lab", 0)
+	if len(got) != 2 || got[0].Text != "do it" || got[1].RefersTo != "lab-1" || got[1].ID != 2 {
+		t.Fatalf("round trip lost data: %+v", got)
+	}
+	if _, err := opened.Append(Message{NestID: "lab", Kind: "deliver", Author: "Munin"}); err != nil {
+		t.Fatal(err)
+	}
+	reopened, err := LoadMessageStore(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n := len(reopened.After("lab", 0)); n != 3 {
+		t.Fatalf("append after reload not persisted: %d messages", n)
+	}
+}