package ingest
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"craftdesign.group/quarry/internal/events"
)
// Forget erases a deleted repository's pipeline state: its publisher
// cursor and publish record, and every node's durable consumer on the
// store event stream with that consumer's fold mark, this node's mark
// included even before it has a consumer. A repository added again
// under the same name, onto an index that was deleted with the old one,
// must republish its census; a cursor that outlived the index would
// resume past it (quarry#294).
//
// All of it is keyed by repository name, and one name may continue
// another's: "seen.indexer-foo-bar" is foo-bar's mark, and so is
// "seen.indexer-foo-<node hash>" when a repository is literally named
// that. Another node's consumer is therefore recognised by the subjects
// it reads, never by its name.
//
// The store events still on the stream go last. A consumer made for the
// next repository of this name reads the stream from its start and
// would fold them into the new index as its history. Names may contain
// dots, so the purge takes exactly one token after the name: foo's
// kinds, never foo.x's subjects.
func Forget(ev *events.Server, repo, node string) error {
if ev == nil || repo == "" {
return nil
}
kv, err := ev.Cursors()
if err != nil {
return fmt.Errorf("ingest: forget %s: %w", repo, err)
}
consumers, err := storeConsumers(ev, repo)
if err != nil {
return fmt.Errorf("ingest: forget %s: %w", repo, err)
}
doomed := []string{repo, "pub." + repo, "seen." + durableName(repo, ""), "seen." + durableName(repo, node)}
for _, name := range consumers {
doomed = append(doomed, "seen."+name)
}
for _, k := range doomed {
if err := kv.Delete(k); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
return fmt.Errorf("ingest: forget %s: %w", repo, err)
}
}
for _, name := range consumers {
if err := ev.JS().DeleteConsumer(events.StreamName, name); err != nil && !errors.Is(err, nats.ErrConsumerNotFound) {
return fmt.Errorf("ingest: forget %s: %w", repo, err)
}
}
purge := &nats.StreamPurgeRequest{Subject: (&Pipeline{code: repo}).subject("*")}
if err := ev.JS().PurgeStream(events.StreamName, purge); err != nil && !errors.Is(err, nats.ErrStreamNotFound) {
return fmt.Errorf("ingest: forget %s: %w", repo, err)
}
return nil
}
// storeConsumers names the pipelines' durable consumers that read
// repo's store events, one per node that has folded it, whether made
// with the exact filter or an older binary's zit.evt.<repo>.>.
func storeConsumers(ev *events.Server, repo string) ([]string, error) {
js, err := jetstream.New(ev.Conn())
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
stream, err := js.Stream(ctx, events.StreamName)
if err != nil {
if errors.Is(err, jetstream.ErrStreamNotFound) {
return nil, nil
}
return nil, err
}
p := &Pipeline{code: repo}
var names []string
list := stream.ListConsumers(ctx)
for c := range list.Info() {
f := c.Config.FilterSubject
if strings.HasPrefix(c.Name, "indexer-") && (f == p.subject("*") || f == p.subject(">")) {
names = append(names, c.Name)
}
}
return names, list.Err()
}