Files
next.orly.dev/pkg/database/save-event.go
2025-10-07 18:18:36 +01:00

232 lines
6.0 KiB
Go

package database
import (
"bytes"
"context"
"errors"
"fmt"
"strings"
"github.com/dgraph-io/badger/v4"
"lol.mleku.dev/chk"
"lol.mleku.dev/log"
"next.orly.dev/pkg/database/indexes"
"next.orly.dev/pkg/database/indexes/types"
"next.orly.dev/pkg/encoders/event"
"next.orly.dev/pkg/encoders/filter"
"next.orly.dev/pkg/encoders/hex"
"next.orly.dev/pkg/encoders/kind"
"next.orly.dev/pkg/encoders/tag"
)
var (
// ErrOlderThanExisting is returned when a candidate event is older than an existing replaceable/addressable event.
ErrOlderThanExisting = errors.New("older than existing event")
// ErrMissingDTag is returned when a parameterized replaceable event lacks the required 'd' tag.
ErrMissingDTag = errors.New("event is missing a d tag identifier")
)
func (d *D) GetSerialsFromFilter(f *filter.F) (
sers types.Uint40s, err error,
) {
var idxs []Range
if idxs, err = GetIndexesFromFilter(f); chk.E(err) {
return
}
for _, idx := range idxs {
var s types.Uint40s
if s, err = d.GetSerialsByRange(idx); chk.E(err) {
continue
}
sers = append(sers, s...)
}
return
}
// WouldReplaceEvent checks if the provided event would replace existing events
// based on Nostr's replaceable or parameterized replaceable semantics. It
// returns true along with the serials of events that should be replaced if the
// candidate is newer-or-equal. If an existing event is newer, it returns
// (false, serials, ErrOlderThanExisting). If no conflicts exist, it returns
// (false, nil, nil).
func (d *D) WouldReplaceEvent(ev *event.E) (bool, types.Uint40s, error) {
// Only relevant for replaceable or parameterized replaceable kinds
if !(kind.IsReplaceable(ev.Kind) || kind.IsParameterizedReplaceable(ev.Kind)) {
return false, nil, nil
}
var f *filter.F
if kind.IsReplaceable(ev.Kind) {
f = &filter.F{
Authors: tag.NewFromBytesSlice(ev.Pubkey),
Kinds: kind.NewS(kind.New(ev.Kind)),
}
} else {
// parameterized replaceable requires 'd' tag
dTag := ev.Tags.GetFirst([]byte("d"))
if dTag == nil {
return false, nil, ErrMissingDTag
}
f = &filter.F{
Authors: tag.NewFromBytesSlice(ev.Pubkey),
Kinds: kind.NewS(kind.New(ev.Kind)),
Tags: tag.NewS(
tag.NewFromAny("d", dTag.Value()),
),
}
}
sers, err := d.GetSerialsFromFilter(f)
if chk.E(err) {
return false, nil, err
}
if len(sers) == 0 {
return false, nil, nil
}
// Determine if any existing event is newer than the candidate
shouldReplace := true
for _, s := range sers {
oldEv, ferr := d.FetchEventBySerial(s)
if chk.E(ferr) {
continue
}
if ev.CreatedAt < oldEv.CreatedAt {
shouldReplace = false
break
}
}
if shouldReplace {
return true, sers, nil
}
return false, sers, ErrOlderThanExisting
}
// SaveEvent saves an event to the database, generating all the necessary indexes.
func (d *D) SaveEvent(c context.Context, ev *event.E) (kc, vc int, err error) {
if ev == nil {
err = errors.New("nil event")
return
}
// check if the event already exists
var ser *types.Uint40
if ser, err = d.GetSerialById(ev.ID); err == nil && ser != nil {
err = errors.New("blocked: event already exists")
return
}
// If the error is "id not found", we can proceed with saving the event
if err != nil && strings.Contains(err.Error(), "id not found in database") {
// Reset error since this is expected for new events
err = nil
} else if err != nil {
// For any other error, return it
// log.E.F("error checking if event exists: %s", err)
return
}
// Check if the event has been deleted before allowing resubmission
if err = d.CheckForDeleted(ev, nil); err != nil {
// log.I.F(
// "SaveEvent: rejecting resubmission of deleted event ID=%s: %v",
// hex.Enc(ev.ID), err,
// )
err = fmt.Errorf("blocked: %s", err.Error())
return
}
// check for replacement (separated check vs deletion)
if kind.IsReplaceable(ev.Kind) || kind.IsParameterizedReplaceable(ev.Kind) {
var wouldReplace bool
var sers types.Uint40s
var werr error
if wouldReplace, sers, werr = d.WouldReplaceEvent(ev); werr != nil {
if errors.Is(werr, ErrOlderThanExisting) {
if kind.IsReplaceable(ev.Kind) {
err = errors.New("blocked: event is older than existing replaceable event")
} else {
err = errors.New("blocked: event is older than existing addressable event")
}
return
}
if errors.Is(werr, ErrMissingDTag) {
// keep behavior consistent with previous implementation
err = ErrMissingDTag
return
}
// any other error
return
}
if wouldReplace {
for _, s := range sers {
var oldEv *event.E
if oldEv, err = d.FetchEventBySerial(s); chk.E(err) {
continue
}
if err = d.DeleteEventBySerial(c, s, oldEv); chk.E(err) {
continue
}
}
}
}
// Get the next sequence number for the event
var serial uint64
if serial, err = d.seq.Next(); chk.E(err) {
return
}
// Generate all indexes for the event
var idxs [][]byte
if idxs, err = GetIndexesForEvent(ev, serial); chk.E(err) {
return
}
// log.I.S(idxs)
for _, k := range idxs {
kc += len(k)
}
// Start a transaction to save the event and all its indexes
err = d.Update(
func(txn *badger.Txn) (err error) {
// Save each index
for _, key := range idxs {
if err = func() (err error) {
// Save the index to the database
if err = txn.Set(key, nil); chk.E(err) {
return err
}
return
}(); chk.E(err) {
return
}
}
// write the event
k := new(bytes.Buffer)
ser := new(types.Uint40)
if err = ser.Set(serial); chk.E(err) {
return
}
if err = indexes.EventEnc(ser).MarshalWrite(k); chk.E(err) {
return
}
v := new(bytes.Buffer)
ev.MarshalBinary(v)
kb, vb := k.Bytes(), v.Bytes()
kc += len(kb)
vc += len(vb)
// log.I.S(kb, vb)
if err = txn.Set(kb, vb); chk.E(err) {
return
}
return
},
)
log.T.F(
"total data written: %d bytes keys %d bytes values for event ID %s", kc,
vc, hex.Enc(ev.ID),
)
// log.T.C(
// func() string {
// return fmt.Sprintf("event:\n%s\n", ev.Serialize())
// },
// )
return
}