Skip to content
quarry Artifact [c64c894223]
Login

Artifact [c64c894223]

Check-ins Using Download Hex

Artifact c64c89422398bc6e4f7fd73688e41469701139faba19eeb9ce62d106fe54a0ce:

// Package chatarchive is the 0001 §5 durability half of chat: it folds a
// channel's live messages into forum-thread artifacts minted through zit,
// so conversation history clones with the repository (the chat window is
// only interim durability). Batching (per N messages or T minutes) keeps
// the DAG from one-artifact-per-line; the archive thread is one forum
// thread per channel, each batch a reply carrying the window's
// transcript. Minting is a store write, so it runs only on the node
// holding the write lease — other nodes drop their buffers, since the
// holder's minted artifact folds back to every node's index.
package chatarchive

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	"craftdesign.group/quarry/internal/chat"
)

// archiveUser authors the forum artifacts. A batch is multi-author, so
// the real speakers are preserved inside the transcript body and the U
// card names the archiver itself.
const archiveUser = "chat-archive"

// threadTitle is the H card of every channel's archive thread.
const threadTitle = "Chat archive"

// Poster mints forum artifacts (zitcli.Client satisfies it).
type Poster interface {
	ForumPost(title, user string, body []byte) (hash string, err error)
	ForumReply(root, user string, body []byte) (hash string, err error)
}

// RootStore persists a channel's archive-thread root across flushes and
// nodes (a NATS KV bucket in production).
type RootStore interface {
	Get(code string) (string, bool)
	Set(code, root string) error
}

// Guard runs a mint under the store's write lease. ran is false when
// this node does not hold the lease (the mint was skipped, not failed);
// err carries a real mint failure when ran is true.
type Guard func(mint func() error) (ran bool, err error)

// Archiver batches one channel's messages and flushes them to the DAG.
type Archiver struct {
	code      string
	sub       func() (<-chan chat.Message, func())
	post      Poster
	roots     RootStore
	guard     Guard
	every     time.Duration
	batchSize int
	buf       []chat.Message
}

// New builds an archiver. every/batchSize bound a batch (0001 §5's
// 15-minute default); a flush fires on whichever comes first.
func New(code string, sub func() (<-chan chat.Message, func()), post Poster, roots RootStore, guard Guard, every time.Duration, batchSize int) *Archiver {
	if every <= 0 {
		every = 15 * time.Minute
	}
	if batchSize <= 0 {
		batchSize = 50
	}
	return &Archiver{code: code, sub: sub, post: post, roots: roots, guard: guard, every: every, batchSize: batchSize}
}

// Run subscribes and archives until ctx is cancelled, flushing whatever
// remains on the way out.
func (a *Archiver) Run(ctx context.Context) {
	msgs, cancel := a.sub()
	defer cancel()
	tick := time.NewTicker(a.every)
	defer tick.Stop()
	for {
		select {
		case <-ctx.Done():
			a.flush()
			return
		case <-tick.C:
			a.flush()
		case m, open := <-msgs:
			if !open {
				a.flush()
				return
			}
			a.buf = append(a.buf, m)
			if len(a.buf) >= a.batchSize {
				a.flush()
			}
		}
	}
}

// flush mints the buffered window as one forum post. The buffer clears
// whether or not this node minted: a non-holder drops it because the
// holder's archive folds back to every node.
func (a *Archiver) flush() {
	if len(a.buf) == 0 {
		return
	}
	body := []byte(transcript(a.buf))
	a.buf = nil
	ran, err := a.guard(func() error {
		if root, ok := a.roots.Get(a.code); ok {
			_, rerr := a.post.ForumReply(root, archiveUser, body)
			return rerr
		}
		h, perr := a.post.ForumPost(threadTitle, archiveUser, body)
		if perr != nil {
			return perr
		}
		return a.roots.Set(a.code, h)
	})
	if ran && err != nil {
		log.Printf("chatarchive %s: %v", a.code, err)
	}
}

// transcript renders a batch as the forum post body: one markdown line
// per message, speakers in bold, so the multi-author window reads as a
// conversation.
func transcript(batch []chat.Message) string {
	var b strings.Builder
	for _, m := range batch {
		fmt.Fprintf(&b, "**%s** %s — %s\n\n", m.User, m.At, m.Text)
	}
	return b.String()
}