Bump version to v0.2.1

app/handle-auth.go

@@ -14,6 +14,11 @@ func (l *Listener) HandleAuth(b []byte) (err error) {
 	if rem, err = env.Unmarshal(b); chk.E(err) {
 		return
 	}
+	defer func() {
+		if env != nil && env.Event != nil {
+			env.Event.Free()
+		}
+	}()
 	if len(rem) > 0 {
 		log.I.F("extra '%s'", rem)
 	}

app/handle-event.go

@@ -21,6 +21,11 @@ func (l *Listener) HandleEvent(msg []byte) (err error) {
 	if msg, err = env.Unmarshal(msg); chk.E(err) {
 		return
 	}
+	defer func() {
+		if env != nil && env.E != nil {
+			env.E.Free()
+		}
+	}()
 	if len(msg) > 0 {
 		log.I.F("extra '%s'", msg)
 	}
@@ -56,21 +61,13 @@ func (l *Listener) HandleEvent(msg []byte) (err error) {
 		}
 		return
 	}
-	// // send a challenge to the client to auth if an ACL is active and not authed
-	// if acl.Registry.Active.Load() != "none" && l.authedPubkey.Load() == nil {
-	// 	log.D.F("sending challenge to %s", l.remote)
-	// 	if err = authenvelope.NewChallengeWith(l.challenge.Load()).
-	// 		Write(l); chk.E(err) {
-	// 		// return
-	// 	}
-	// 	// ACL is enabled so return and wait for auth
-	// 	// return
-	// }
 	// check permissions of user
 	accessLevel := acl.Registry.GetAccessLevel(l.authedPubkey.Load())
 	switch accessLevel {
 	case "none":
-		log.D.F("handle event: sending CLOSED to %s", l.remote)
+		log.D.F(
+			"handle event: sending 'OK,false,auth-required...' to %s", l.remote,
+		)
 		if err = okenvelope.NewFrom(
 			env.Id(), false,
 			reason.AuthRequired.F("auth required for write access"),
@@ -84,17 +81,20 @@ func (l *Listener) HandleEvent(msg []byte) (err error) {
 		}
 		return
 	case "read":
-		log.D.F("handle event: sending CLOSED to %s", l.remote)
+		log.D.F(
+			"handle event: sending 'OK,false,auth-required:...' to %s",
+			l.remote,
+		)
 		if err = okenvelope.NewFrom(
 			env.Id(), false,
 			reason.AuthRequired.F("auth required for write access"),
 		).Write(l); chk.E(err) {
-			// return
+			return
 		}
 		log.D.F("handle event: sending challenge to %s", l.remote)
 		if err = authenvelope.NewChallengeWith(l.challenge.Load()).
 			Write(l); chk.E(err) {
-			// return
+			return
 		}
 		return
 	default:
@@ -122,16 +122,26 @@ func (l *Listener) HandleEvent(msg []byte) (err error) {
 	if _, _, err = l.SaveEvent(l.Ctx, env.E); chk.E(err) {
 		return
 	}
-	// if a follow list was saved, reconfigure ACLs now that it is persisted
-	if env.E.Kind == kind.FollowList.K {
-		if err = acl.Registry.Configure(); chk.E(err) {
-		}
-	}
-	l.publishers.Deliver(env.E)
 	// Send a success response storing
 	if err = Ok.Ok(l, env, ""); chk.E(err) {
 		return
 	}
+	defer l.publishers.Deliver(env.E)
 	log.D.F("saved event %0x", env.E.ID)
+	var isNewFromAdmin bool
+	for _, admin := range l.Admins {
+		if utils.FastEqual(admin, env.E.Pubkey) {
+			isNewFromAdmin = true
+			break
+		}
+	}
+	if isNewFromAdmin {
+		// if a follow list was saved, reconfigure ACLs now that it is persisted
+		if env.E.Kind == kind.FollowList.K ||
+			env.E.Kind == kind.RelayListMetadata.K {
+			if err = acl.Registry.Configure(); chk.E(err) {
+			}
+		}
+	}
 	return
 }

app/handle-relayinfo.go

@@ -33,32 +33,32 @@ func (s *Server) HandleRelayInfo(w http.ResponseWriter, r *http.Request) {
 		relayinfo.BasicProtocol,
 		// relayinfo.Authentication,
 		// relayinfo.EncryptedDirectMessage,
-		// relayinfo.EventDeletion,
+		relayinfo.EventDeletion,
 		relayinfo.RelayInformationDocument,
 		// relayinfo.GenericTagQueries,
 		// relayinfo.NostrMarketplace,
-		// relayinfo.EventTreatment,
+		relayinfo.EventTreatment,
 		// relayinfo.CommandResults,
-		// relayinfo.ParameterizedReplaceableEvents,
+		relayinfo.ParameterizedReplaceableEvents,
 		// relayinfo.ExpirationTimestamp,
-		// relayinfo.ProtectedEvents,
-		// relayinfo.RelayListMetadata,
+		relayinfo.ProtectedEvents,
+		relayinfo.RelayListMetadata,
 	)
 	if s.Config.ACLMode != "none" {
 		supportedNIPs = relayinfo.GetList(
 			relayinfo.BasicProtocol,
 			relayinfo.Authentication,
 			// relayinfo.EncryptedDirectMessage,
-			// relayinfo.EventDeletion,
+			relayinfo.EventDeletion,
 			relayinfo.RelayInformationDocument,
 			// relayinfo.GenericTagQueries,
 			// relayinfo.NostrMarketplace,
-			// relayinfo.EventTreatment,
+			relayinfo.EventTreatment,
 			// relayinfo.CommandResults,
 			// relayinfo.ParameterizedReplaceableEvents,
 			// relayinfo.ExpirationTimestamp,
-			// relayinfo.ProtectedEvents,
-			// relayinfo.RelayListMetadata,
+			relayinfo.ProtectedEvents,
+			relayinfo.RelayListMetadata,
 		)
 	}
 	sort.Sort(supportedNIPs)

app/handle-req.go

@@ -12,11 +12,14 @@ import (
 	"encoders.orly/envelopes/reqenvelope"
 	"encoders.orly/event"
 	"encoders.orly/filter"
+	"encoders.orly/hex"
+	"encoders.orly/kind"
 	"encoders.orly/reason"
 	"encoders.orly/tag"
 	"github.com/dgraph-io/badger/v4"
 	"lol.mleku.dev/chk"
 	"lol.mleku.dev/log"
+	utils "utils.orly"
 	"utils.orly/normalize"
 	"utils.orly/pointers"
 )
@@ -32,33 +35,11 @@ func (l *Listener) HandleReq(msg []byte) (
 	if len(rem) > 0 {
 		log.I.F("extra '%s'", rem)
 	}
-	// // send a challenge to the client to auth if an ACL is active and not authed
-	// if acl.Registry.Active.Load() != "none" && l.authedPubkey.Load() == nil {
-	// 	log.D.F("sending challenge to %s", l.remote)
-	// 	if err = authenvelope.NewChallengeWith(l.challenge.Load()).
-	// 		Write(l); chk.E(err) {
-	// 		// return
-	// 	}
-	// 	log.D.F("sending CLOSED to %s", l.remote)
-	// 	if err = closedenvelope.NewFrom(
-	// 		env.Subscription, reason.AuthRequired.F("auth required for access"),
-	// 	).Write(l); chk.E(err) {
-	// 		return
-	// 	}
-	// 	// ACL is enabled so return and wait for auth
-	// 	// return
-	// }
 	// send a challenge to the client to auth if an ACL is active
 	if acl.Registry.Active.Load() != "none" {
-		// log.D.F("sending CLOSED to %s", l.remote)
-		// if err = closedenvelope.NewFrom(
-		// 	env.Subscription, reason.AuthRequired.F("auth required for access"),
-		// ).Write(l); chk.E(err) {
-		// 	// return
-		// }
 		if err = authenvelope.NewChallengeWith(l.challenge.Load()).
 			Write(l); chk.E(err) {
-			// return
+			return
 		}
 	}
 	// check permissions of user
@@ -90,7 +71,48 @@ func (l *Listener) HandleReq(msg []byte) (
 			err = nil
 		}
 	}
-	// write out the events to the socket
+	var tmp event.S
+privCheck:
+	for _, ev := range events {
+		if kind.IsPrivileged(ev.Kind) &&
+			accessLevel != "admin" { // admins can see all events
+			log.I.F("checking privileged event %s", ev.ID)
+			pk := l.authedPubkey.Load()
+			if pk == nil {
+				continue
+			}
+			if utils.FastEqual(ev.Pubkey, pk) {
+				log.I.F(
+					"privileged event %s is for logged in pubkey %0x", ev.ID,
+					pk,
+				)
+				tmp = append(tmp, ev)
+				continue
+			}
+			pTags := ev.Tags.GetAll([]byte("p"))
+			for _, pTag := range pTags {
+				var pt []byte
+				if pt, err = hex.Dec(string(pTag.Value())); chk.E(err) {
+					continue
+				}
+				if utils.FastEqual(pt, pk) {
+					log.I.F(
+						"privileged event %s is for logged in pubkey %0x",
+						ev.ID, pk,
+					)
+					tmp = append(tmp, ev)
+					continue privCheck
+				}
+			}
+			log.W.F(
+				"privileged event %s does not contain the logged in pubkey %0x",
+				ev.ID, pk,
+			)
+		} else {
+			tmp = append(tmp, ev)
+		}
+	}
+	events = tmp
 	seen := make(map[string]struct{})
 	for _, ev := range events {
 		// track the IDs we've sent
@@ -150,11 +172,12 @@ func (l *Listener) HandleReq(msg []byte) (
 	if !cancel {
 		l.publishers.Receive(
 			&W{
-				Conn:     l.conn,
-				remote:   l.remote,
-				Id:       string(env.Subscription),
-				Receiver: receiver,
-				Filters:  env.Filters,
+				Conn:         l.conn,
+				remote:       l.remote,
+				Id:           string(env.Subscription),
+				Receiver:     receiver,
+				Filters:      env.Filters,
+				AuthedPubkey: l.authedPubkey.Load(),
 			},
 		)
 	} else {

app/main.go

@@ -34,7 +34,6 @@ func Run(
 		}
 		adminKeys = append(adminKeys, pk)
 	}
-
 	// start listener
 	l := &Server{
 		Ctx:        ctx,

app/publisher.go

@@ -8,17 +8,21 @@ import (
 	"encoders.orly/envelopes/eventenvelope"
 	"encoders.orly/event"
 	"encoders.orly/filter"
+	"encoders.orly/hex"
+	"encoders.orly/kind"
 	"github.com/coder/websocket"
 	"interfaces.orly/publisher"
 	"interfaces.orly/typer"
 	"lol.mleku.dev/chk"
 	"lol.mleku.dev/log"
+	utils "utils.orly"
 )
 
 const Type = "socketapi"
 
 type Subscription struct {
-	remote string
+	remote       string
+	AuthedPubkey []byte
 	*filter.S
 }
 
@@ -46,6 +50,9 @@ type W struct {
 	// associated with this WebSocket connection. It is used to determine which
 	// notifications or data should be received by the subscriber.
 	Filters *filter.S
+
+	// AuthedPubkey is the authenticated pubkey associated with the listener (if any).
+	AuthedPubkey []byte
 }
 
 func (w *W) Type() (typeName string) { return Type }
@@ -56,7 +63,7 @@ func (w *W) Type() (typeName string) { return Type }
 type P struct {
 	c context.Context
 	// Mx is the mutex for the Map.
-	Mx sync.Mutex
+	Mx sync.RWMutex
 	// Map is the map of subscribers and subscriptions from the websocket api.
 	Map
 }
@@ -112,7 +119,7 @@ func (p *P) Receive(msg typer.T) {
 		defer p.Mx.Unlock()
 		if subs, ok := p.Map[m.Conn]; !ok {
 			subs = make(map[string]Subscription)
-			subs[m.Id] = Subscription{S: m.Filters, remote: m.remote}
+			subs[m.Id] = Subscription{S: m.Filters, remote: m.remote, AuthedPubkey: m.AuthedPubkey}
 			p.Map[m.Conn] = subs
 			log.D.C(
 				func() string {
@@ -124,7 +131,7 @@ func (p *P) Receive(msg typer.T) {
 				},
 			)
 		} else {
-			subs[m.Id] = Subscription{S: m.Filters, remote: m.remote}
+			subs[m.Id] = Subscription{S: m.Filters, remote: m.remote, AuthedPubkey: m.AuthedPubkey}
 			log.D.C(
 				func() string {
 					return fmt.Sprintf(
@@ -150,48 +157,77 @@ func (p *P) Receive(msg typer.T) {
 // for unauthenticated users when events are privileged.
 func (p *P) Deliver(ev *event.E) {
 	var err error
-	p.Mx.Lock()
-	defer p.Mx.Unlock()
+	// Snapshot the deliveries under read lock to avoid holding locks during I/O
+	p.Mx.RLock()
+	type delivery struct {
+		w   *websocket.Conn
+		id  string
+		sub Subscription
+	}
+	var deliveries []delivery
+	for w, subs := range p.Map {
+		for id, subscriber := range subs {
+			if subscriber.Match(ev) {
+				deliveries = append(deliveries, delivery{w: w, id: id, sub: subscriber})
+			}
+		}
+	}
+	p.Mx.RUnlock()
 	log.D.C(
 		func() string {
 			return fmt.Sprintf(
 				"delivering event %0x to websocket subscribers %d", ev.ID,
-				len(p.Map),
+				len(deliveries),
 			)
 		},
 	)
-	for w, subs := range p.Map {
-		for id, subscriber := range subs {
-			if !subscriber.Match(ev) {
-				continue
-			}
-			// if p.Server.AuthRequired() {
-			// 	if !auth.CheckPrivilege(w.AuthedPubkey(), ev) {
-			// 		continue
-			// 	}
-			// }
-			var res *eventenvelope.Result
-			if res, err = eventenvelope.NewResultWith(id, ev); chk.E(err) {
-				continue
-			}
-			if err = w.Write(
-				p.c, websocket.MessageText, res.Marshal(nil),
-			); chk.E(err) {
-				p.removeSubscriber(w)
-				if err = w.CloseNow(); chk.E(err) {
-					continue
+	for _, d := range deliveries {
+		// If the event is privileged, enforce that the subscriber's authed pubkey matches
+		// either the event pubkey or appears in any 'p' tag of the event.
+		if kind.IsPrivileged(ev.Kind) && len(d.sub.AuthedPubkey) > 0 {
+			pk := d.sub.AuthedPubkey
+			allowed := false
+			// Direct author match
+			if utils.FastEqual(ev.Pubkey, pk) {
+				allowed = true
+			} else if ev.Tags != nil {
+				for _, pTag := range ev.Tags.GetAll([]byte("p")) {
+					// pTag.Value() returns []byte hex string; decode to bytes
+					dec, derr := hex.Dec(string(pTag.Value()))
+					if derr != nil {
+						continue
+					}
+					if utils.FastEqual(dec, pk) {
+						allowed = true
+						break
+					}
 				}
+			}
+			if !allowed {
+				// Skip delivery for this subscriber
 				continue
 			}
-			log.D.C(
-				func() string {
-					return fmt.Sprintf(
-						"dispatched event %0x to subscription %s, %s",
-						ev.ID, id, subscriber.remote,
-					)
-				},
-			)
 		}
+		var res *eventenvelope.Result
+		if res, err = eventenvelope.NewResultWith(d.id, ev); chk.E(err) {
+			continue
+		}
+		if err = d.w.Write(
+			p.c, websocket.MessageText, res.Marshal(nil),
+		); chk.E(err) {
+			// On error, remove the subscriber connection safely
+			p.removeSubscriber(d.w)
+			_ = d.w.CloseNow()
+			continue
+		}
+		log.D.C(
+			func() string {
+				return fmt.Sprintf(
+					"dispatched event %0x to subscription %s, %s",
+					ev.ID, d.id, d.sub.remote,
+				)
+			},
+		)
 	}
 }
 
@@ -199,6 +235,7 @@ func (p *P) Deliver(ev *event.E) {
 // websocket.
 func (p *P) removeSubscriberId(ws *websocket.Conn, id string) {
 	p.Mx.Lock()
+	defer p.Mx.Unlock()
 	var subs map[string]Subscription
 	var ok bool
 	if subs, ok = p.Map[ws]; ok {
@@ -208,13 +245,12 @@ func (p *P) removeSubscriberId(ws *websocket.Conn, id string) {
 			delete(p.Map, ws)
 		}
 	}
-	p.Mx.Unlock()
 }
 
 // removeSubscriber removes a websocket from the P collection.
 func (p *P) removeSubscriber(ws *websocket.Conn) {
 	p.Mx.Lock()
+	defer p.Mx.Unlock()
 	clear(p.Map[ws])
 	delete(p.Map, ws)
-	p.Mx.Unlock()
 }

main.go

@@ -30,9 +30,10 @@ func main() {
 		os.Exit(1)
 	}
 	acl.Registry.Active.Store(cfg.ACLMode)
-	if err = acl.Registry.Configure(cfg, db); chk.E(err) {
+	if err = acl.Registry.Configure(cfg, db, ctx); chk.E(err) {
 		os.Exit(1)
 	}
+	acl.Registry.Syncer()
 	quit := app.Run(ctx, cfg, db)
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, os.Interrupt)

pkg/acl/acl.go

@@ -48,6 +48,15 @@ func (s *S) GetACLInfo() (name, description, documentation string) {
 	return
 }
 
+func (s *S) Syncer() {
+	for _, i := range s.ACL {
+		if i.Type() == s.Active.Load() {
+			i.Syncer()
+			break
+		}
+	}
+}
+
 func (s *S) Type() (typ string) {
 	for _, i := range s.ACL {
 		if i.Type() == s.Active.Load() {

pkg/acl/follows.go

@@ -1,30 +1,43 @@
 package acl
 
 import (
+	"context"
 	"reflect"
+	"strings"
 	"sync"
+	"time"
 
 	database "database.orly"
 	"database.orly/indexes/types"
 	"encoders.orly/bech32encoding"
+	"encoders.orly/envelopes"
+	"encoders.orly/envelopes/eoseenvelope"
+	"encoders.orly/envelopes/eventenvelope"
+	"encoders.orly/envelopes/reqenvelope"
 	"encoders.orly/event"
 	"encoders.orly/filter"
 	"encoders.orly/hex"
 	"encoders.orly/kind"
 	"encoders.orly/tag"
+	"github.com/coder/websocket"
 	"lol.mleku.dev/chk"
 	"lol.mleku.dev/errorf"
 	"lol.mleku.dev/log"
 	"next.orly.dev/app/config"
 	utils "utils.orly"
+	"utils.orly/normalize"
+	"utils.orly/values"
 )
 
 type Follows struct {
+	Ctx context.Context
 	cfg *config.C
 	*database.D
-	followsMx sync.RWMutex
-	admins    [][]byte
-	follows   [][]byte
+	followsMx  sync.RWMutex
+	admins     [][]byte
+	follows    [][]byte
+	updated    chan struct{}
+	subsCancel context.CancelFunc
 }
 
 func (f *Follows) Configure(cfg ...any) (err error) {
@@ -37,6 +50,9 @@ func (f *Follows) Configure(cfg ...any) (err error) {
 		case *database.D:
 			log.D.F("setting ACL database: %s", c.Path())
 			f.D = c
+		case context.Context:
+			log.D.F("setting ACL context: %s", c.Value("id"))
+			f.Ctx = c
 		default:
 			err = errorf.E("invalid type: %T", reflect.TypeOf(ca))
 		}
@@ -92,6 +108,11 @@ func (f *Follows) Configure(cfg ...any) (err error) {
 			}
 		}
 	}
+	if f.updated == nil {
+		f.updated = make(chan struct{})
+	} else {
+		f.updated <- struct{}{}
+	}
 	return
 }
 
@@ -121,6 +142,199 @@ func (f *Follows) GetACLInfo() (name, description, documentation string) {
 
 func (f *Follows) Type() string { return "follows" }
 
+func (f *Follows) adminRelays() (urls []string) {
+	f.followsMx.RLock()
+	admins := make([][]byte, len(f.admins))
+	copy(admins, f.admins)
+	f.followsMx.RUnlock()
+	seen := make(map[string]struct{})
+	for _, adm := range admins {
+		fl := &filter.F{
+			Authors: tag.NewFromAny(adm),
+			Kinds:   kind.NewS(kind.New(kind.RelayListMetadata.K)),
+		}
+		idxs, err := database.GetIndexesFromFilter(fl)
+		if chk.E(err) {
+			continue
+		}
+		var sers types.Uint40s
+		for _, idx := range idxs {
+			s, err := f.D.GetSerialsByRange(idx)
+			if chk.E(err) {
+				continue
+			}
+			sers = append(sers, s...)
+		}
+		for _, s := range sers {
+			ev, err := f.D.FetchEventBySerial(s)
+			if chk.E(err) || ev == nil {
+				continue
+			}
+			for _, v := range ev.Tags.GetAll([]byte("r")) {
+				u := string(v.Value())
+				n := string(normalize.URL(u))
+				if n == "" {
+					continue
+				}
+				if _, ok := seen[n]; ok {
+					continue
+				}
+				seen[n] = struct{}{}
+				urls = append(urls, n)
+			}
+		}
+	}
+	return
+}
+
+func (f *Follows) startSubscriptions(ctx context.Context) {
+	// build authors list: admins + follows
+	f.followsMx.RLock()
+	authors := make([][]byte, 0, len(f.admins)+len(f.follows))
+	authors = append(authors, f.admins...)
+	authors = append(authors, f.follows...)
+	f.followsMx.RUnlock()
+	if len(authors) == 0 {
+		log.W.F("follows syncer: no authors (admins+follows) to subscribe to")
+		return
+	}
+	urls := f.adminRelays()
+	if len(urls) == 0 {
+		log.W.F("follows syncer: no admin relays found in DB (kind 10002)")
+		return
+	}
+	log.I.F(
+		"follows syncer: subscribing to %d relays for %d authors", len(urls),
+		len(authors),
+	)
+	for _, u := range urls {
+		u := u
+		go func() {
+			backoff := time.Second
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+				c, _, err := websocket.Dial(ctx, u, nil)
+				if err != nil {
+					log.W.F("follows syncer: dial %s failed: %v", u, err)
+					timer := time.NewTimer(backoff)
+					select {
+					case <-ctx.Done():
+						return
+					case <-timer.C:
+					}
+					if backoff < 30*time.Second {
+						backoff *= 2
+					}
+					continue
+				}
+				backoff = time.Second
+				// send REQ
+				ff := &filter.S{}
+				f1 := &filter.F{
+					Authors: tag.NewFromBytesSlice(authors...),
+					Limit:   values.ToUintPointer(0),
+				}
+				*ff = append(*ff, f1)
+				req := reqenvelope.NewFrom([]byte("follows-sync"), ff)
+				if err := c.Write(
+					ctx, websocket.MessageText, req.Marshal(nil),
+				); chk.E(err) {
+					_ = c.Close(websocket.StatusInternalError, "write failed")
+					continue
+				}
+				log.I.F("sent REQ to %s for follows subscription", u)
+				// read loop
+				for {
+					select {
+					case <-ctx.Done():
+						_ = c.Close(websocket.StatusNormalClosure, "ctx done")
+						return
+					default:
+					}
+					_, data, err := c.Read(ctx)
+					if err != nil {
+						_ = c.Close(websocket.StatusNormalClosure, "read err")
+						break
+					}
+					label, rem, err := envelopes.Identify(data)
+					if chk.E(err) {
+						continue
+					}
+					switch label {
+					case eventenvelope.L:
+						res, _, err := eventenvelope.ParseResult(rem)
+						if chk.E(err) || res == nil || res.Event == nil {
+							continue
+						}
+						// verify signature before saving
+						if ok, err := res.Event.Verify(); chk.T(err) || !ok {
+							continue
+						}
+						if _, _, err := f.D.SaveEvent(
+							ctx, res.Event,
+						); err != nil {
+							if !strings.HasPrefix(
+								err.Error(), "event already exists",
+							) {
+								log.W.F(
+									"follows syncer: save event failed: %v",
+									err,
+								)
+							}
+							// ignore duplicates and continue
+						}
+						log.I.F(
+							"saved new event from follows syncer: %0x",
+							res.Event.ID,
+						)
+					case eoseenvelope.L:
+						// ignore, continue subscription
+					default:
+						// ignore other labels
+					}
+				}
+				// loop reconnect
+			}
+		}()
+	}
+}
+
+func (f *Follows) Syncer() {
+	log.I.F("starting follows syncer")
+	go func() {
+		// start immediately if Configure already ran
+		for {
+			var innerCancel context.CancelFunc
+			select {
+			case <-f.Ctx.Done():
+				if f.subsCancel != nil {
+					f.subsCancel()
+				}
+				return
+			case <-f.updated:
+				// close and reopen subscriptions to users on the follow list and admins
+				if f.subsCancel != nil {
+					log.I.F("follows syncer: cancelling existing subscriptions")
+					f.subsCancel()
+				}
+				ctx, cancel := context.WithCancel(f.Ctx)
+				f.subsCancel = cancel
+				innerCancel = cancel
+				log.I.F("follows syncer: (re)opening subscriptions")
+				f.startSubscriptions(ctx)
+			}
+			// small sleep to avoid tight loop if updated fires rapidly
+			if innerCancel == nil {
+				time.Sleep(50 * time.Millisecond)
+			}
+		}
+	}()
+}
+
 func init() {
 	log.T.F("registering follows ACL")
 	Registry.Register(new(Follows))

pkg/acl/none.go

@@ -20,6 +20,8 @@ func (n None) Type() string {
 	return "none"
 }
 
+func (n None) Syncer() {}
+
 func init() {
 	log.T.F("registering none ACL")
 	Registry.Register(new(None))

pkg/crypto/p256k/btcec/btcec.go

@@ -15,7 +15,7 @@ import (
 type Signer struct {
 	SecretKey *secp256k1.SecretKey
 	PublicKey *secp256k1.PublicKey
-	BTCECSec  *ec.SecretKey
+	BTCECSec  *secp256k1.SecretKey
 	pkb, skb  []byte
 }
 
@@ -23,11 +23,11 @@ var _ signer.I = &Signer{}
 
 // Generate creates a new Signer.
 func (s *Signer) Generate() (err error) {
-	if s.SecretKey, err = ec.NewSecretKey(); chk.E(err) {
+	if s.SecretKey, err = secp256k1.GenerateSecretKey(); chk.E(err) {
 		return
 	}
 	s.skb = s.SecretKey.Serialize()
-	s.BTCECSec, _ = ec.PrivKeyFromBytes(s.skb)
+	s.BTCECSec = secp256k1.PrivKeyFromBytes(s.skb)
 	s.PublicKey = s.SecretKey.PubKey()
 	s.pkb = schnorr.SerializePubKey(s.PublicKey)
 	return
@@ -43,7 +43,7 @@ func (s *Signer) InitSec(sec []byte) (err error) {
 	s.SecretKey = secp256k1.SecKeyFromBytes(sec)
 	s.PublicKey = s.SecretKey.PubKey()
 	s.pkb = schnorr.SerializePubKey(s.PublicKey)
-	s.BTCECSec, _ = ec.PrivKeyFromBytes(s.skb)
+	s.BTCECSec = secp256k1.PrivKeyFromBytes(s.skb)
 	return
 }
 
@@ -142,7 +142,7 @@ func (s *Signer) ECDH(pubkeyBytes []byte) (secret []byte, err error) {
 	); chk.E(err) {
 		return
 	}
-	secret = ec.GenerateSharedSecret(s.BTCECSec, pub)
+	secret = secp256k1.GenerateSharedSecret(s.BTCECSec, pub)
 	return
 }
 
@@ -154,7 +154,7 @@ type Keygen struct {
 // Generate a new key pair. If the result is suitable, the embedded Signer can have its contents
 // extracted.
 func (k *Keygen) Generate() (pubBytes []byte, err error) {
-	if k.Signer.SecretKey, err = ec.NewSecretKey(); chk.E(err) {
+	if k.Signer.SecretKey, err = secp256k1.GenerateSecretKey(); chk.E(err) {
 		return
 	}
 	k.Signer.PublicKey = k.SecretKey.PubKey()

pkg/database/export.go

@@ -49,6 +49,7 @@ func (d *D) Export(c context.Context, w io.Writer, pubkeys ...[]byte) {
 					if _, err = w.Write([]byte{'\n'}); chk.E(err) {
 						return
 					}
+					ev.Free()
 					evBuf.Reset()
 				}
 				return
@@ -86,14 +87,15 @@ func (d *D) Export(c context.Context, w io.Writer, pubkeys ...[]byte) {
 						if err = ev.UnmarshalBinary(evBuf); chk.E(err) {
 							continue
 						}
-						// Serialize the event to JSON and write it to the output
-						if _, err = w.Write(ev.Serialize()); chk.E(err) {
-							continue
-						}
-						if _, err = w.Write([]byte{'\n'}); chk.E(err) {
-							continue
-						}
-						evBuf.Reset()
+ 					// Serialize the event to JSON and write it to the output
+ 					if _, err = w.Write(ev.Serialize()); chk.E(err) {
+ 						continue
+ 					}
+ 					if _, err = w.Write([]byte{'\n'}); chk.E(err) {
+ 						continue
+ 					}
+ 					ev.Free()
+ 					evBuf.Reset()
 					}
 					return
 				},

pkg/database/import.go

@@ -52,17 +52,22 @@ func (d *D) Import(rr io.Reader) {
 				continue
 			}
 
-			ev := &event.E{}
+			ev := event.New()
 			if _, err = ev.Unmarshal(b); err != nil {
+				// return the pooled buffer on error
+				ev.Free()
 				continue
 			}
 
 			if _, _, err = d.SaveEvent(d.ctx, ev); err != nil {
+				// return the pooled buffer on error paths too
+				ev.Free()
 				continue
 			}
 
+			// return the pooled buffer after successful save
+			ev.Free()
 			b = nil
-			ev = nil
 			count++
 			if count%100 == 0 {
 				log.I.F("received %d events", count)

pkg/encoders/kind/kind.go

@@ -78,9 +78,9 @@ var Privileged = []*K{
 
 // IsPrivileged returns true if the type is the kind of message nobody else than
 // the pubkeys in the event and p tags of the event are party to.
-func (k *K) IsPrivileged() (is bool) {
+func IsPrivileged(k uint16) (is bool) {
 	for i := range Privileged {
-		if k.Equal(Privileged[i].K) {
+		if k == Privileged[i].K {
 			return true
 		}
 	}

pkg/encoders/kind/kinds.go

@@ -142,7 +142,7 @@ func (k *S) Unmarshal(b []byte) (r []byte, err error) {
 // be privacy protected).
 func (k *S) IsPrivileged() (priv bool) {
 	for i := range k.K {
-		if k.K[i].IsPrivileged() {
+		if IsPrivileged(k.K[i].K) {
 			return true
 		}
 	}

pkg/interfaces/acl/acl.go

@@ -27,5 +27,9 @@ type I interface {
 	// explain briefly how it works, and then a long text of documentation of
 	// the ACL's rules and configuration (in asciidoc or markdown).
 	GetACLInfo() (name, description, documentation string)
+	// Syncer is a worker thread that does things in the background like syncing
+	// with other relays on admin relay lists using subscriptions for all events
+	// that arrive elsewhere relevant to the ACL scheme.
+	Syncer()
 	typer.T
 }

pkg/version/version

@@ -1 +1 @@
-v0.1.0
+v0.2.1