package events
import (
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go"
)
// ForgetRepo erases what the streams hold about a deleted repository:
// the attention events addressed to anybody about it, its event log,
// and its traffic tallies. A repository created later under the same
// name must not inherit them (quarry#294) — an inbox row that opens the
// new repository, a clone count and the accounts behind it that were
// somebody else's.
//
// Messages are matched on the repository they NAME, never on a subject
// prefix alone: repository names may contain dots, so "gone.>" also
// covers "gone.x", which is a different repository.
func (s *Server) ForgetRepo(repo string) error {
if s == nil || repo == "" {
return nil
}
named := func(data []byte) bool {
var e struct {
Repo string `json:"repo"`
}
return json.Unmarshal(data, &e) == nil && e.Repo == repo
}
if err := s.deleteWhere(AttnStream, "attn.>", named); err != nil {
return err
}
if err := s.deleteWhere(VocabStream, "quarry.evt."+repo+".>", named); err != nil {
return err
}
// Counter bodies carry only the total, so these match by subject,
// spelled out to the families cmd/quarry counts: a name, then
// nothing, anon, or an auth/claim/client token. The token after the
// family is what tells repo's tallies from those of a repository
// named repo.clone or repo.fetch.
for _, fam := range []string{"clone", "fetch"} {
head := CounterSubject(repo, fam)
for _, subj := range []string{head, head + ".anon", head + ".auth.*",
head + ".claim.*", head + ".client.*.*"} {
if err := s.js.PurgeStream(CounterStream, &nats.StreamPurgeRequest{Subject: subj}); err != nil &&
!errors.Is(err, nats.ErrStreamNotFound) {
return fmt.Errorf("events: forget %s tallies: %w", repo, err)
}
}
}
return nil
}
// deleteWhere deletes every message under filter in stream whose body
// match accepts.
func (s *Server) deleteWhere(stream, filter string, match func([]byte) bool) error {
sub, err := s.js.PullSubscribe(filter, "",
nats.DeliverAll(), nats.AckNone(), nats.BindStream(stream),
nats.InactiveThreshold(30*time.Second))
if err != nil {
if errors.Is(err, nats.ErrStreamNotFound) {
return nil
}
return fmt.Errorf("events: forget in %s: %w", stream, err)
}
defer sub.Unsubscribe()
var doomed []uint64
for {
msgs, ferr := sub.Fetch(256, nats.MaxWait(200*time.Millisecond))
if ferr != nil && !errors.Is(ferr, nats.ErrTimeout) {
return fmt.Errorf("events: forget in %s: %w", stream, ferr)
}
for _, m := range msgs {
if !match(m.Data) {
continue
}
if meta, merr := m.Metadata(); merr == nil {
doomed = append(doomed, meta.Sequence.Stream)
}
}
if len(msgs) < 256 {
break
}
}
for _, seq := range doomed {
if err := s.js.DeleteMsg(stream, seq); err != nil && !errors.Is(err, nats.ErrMsgNotFound) {
return fmt.Errorf("events: forget %s#%d: %w", stream, seq, err)
}
}
return nil
}