feat(membership): single-use invites + hard-delete in the Store (SQLite + KV)

Add the data layer for WhatsApp-style accounts on the wallet model: the admin mints a single-use invitation link, the new user redeems it by publishing only its public keys, and the admin can hard-delete a user. - Invite type and lifecycle (invites.go): 32-byte crypto/rand hex token, 7-day default TTL, fail-closed expiry parsing. Methods CreateInvite/GetInvite/ ListInvites/ConsumeInvite/CancelInvite on both backends. ConsumeInvite is atomic and single-use: SQLite uses a transaction guarded by `used = 0`, the KV store uses a compare-and-swap on the entry revision (mark-first). Both burn the token on claim, so an already-registered key surfaces ErrUserExists with the invite spent — identical semantics across backends. - DeleteUser (users.go + jetstream_store.go): hard-delete of the allowlist row, distinct from RevokeUser's status flip. Room memberships of the ex-user are intentionally left inert (they can no longer authenticate); no partial cleanup. - Migration 003_invites.sql (root + embedded copy, byte-identical): additive `invites` table with audit columns, per db_migrations rules. - Store interface gains DeleteUser, CreateInvite, GetInvite, ListInvites, ConsumeInvite, CancelInvite. New UNIBUS_invites KV bucket. - Consistency fix: SQLite GetUser now maps sql.ErrNoRows to ErrNotFound, matching the KV backend and the storage-agnostic contract documented in store.go. - ValidateKexPubHex added alongside ValidateSignPubHex for /register key checks. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-07 22:14:25 +02:00
parent f31580deec
commit d64b0c052d
6 changed files with 607 additions and 2 deletions
@@ -50,6 +50,7 @@ const (
 	bucketByMember  = "UNIBUS_rooms_by_member"
 	bucketRoomKeys  = "UNIBUS_room_keys"
 	bucketUsers     = "UNIBUS_users"
+	bucketInvites   = "UNIBUS_invites"
 	defaultKVOpTime = 5 * time.Second
 )

@@ -71,6 +72,7 @@ type jetstreamStore struct {
 	byMember  jetstream.KeyValue
 	keys      jetstream.KeyValue
 	users     jetstream.KeyValue
+	invites   jetstream.KeyValue
 	opTimeout time.Duration
 }

@@ -108,6 +110,7 @@ func OpenJetStream(js jetstream.JetStream, cfg JetStreamConfig) (Store, error) {
 		{bucketByMember, &s.byMember},
 		{bucketRoomKeys, &s.keys},
 		{bucketUsers, &s.users},
+		{bucketInvites, &s.invites},
 	} {
 		var kv jetstream.KeyValue
 		var lastErr error
@@ -498,6 +501,28 @@ func (s *jetstreamStore) RevokeUser(signPub string) error {
 	return nil
 }

+// DeleteUser hard-deletes a user from the KV allowlist (the purge counterpart of
+// RevokeUser's status flip). It checks existence first so deleting an unknown key
+// is ErrNotFound (KV Delete is otherwise idempotent and would not signal a miss).
+// Only the allowlist key is removed; room memberships the ex-user holds become
+// inert because they can no longer authenticate — see the SQLite DeleteUser for
+// the full rationale on why room state is left untouched.
+func (s *jetstreamStore) DeleteUser(signPub string) error {
+	signPub = normalizeSignPub(signPub)
+	ctx, cancel := s.ctx()
+	defer cancel()
+	if _, err := s.users.Get(ctx, signPub); err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return fmt.Errorf("membership: delete user %q: %w", signPub, ErrNotFound)
+		}
+		return fmt.Errorf("membership: delete user %q: %w", signPub, err)
+	}
+	if err := s.users.Delete(ctx, signPub); err != nil {
+		return fmt.Errorf("membership: delete user %q: %w", signPub, err)
+	}
+	return nil
+}
+
 // IsAuthorized reports whether signPub is an active bus user. Any backend error
 // (including a KV quorum loss or timeout) yields false: fail closed.
 func (s *jetstreamStore) IsAuthorized(signPub string) bool {
@@ -533,6 +558,173 @@ func (s *jetstreamStore) HasAdmin() bool {
 	return false
 }

+// ---- invites (single-use registration tokens) ----------------------------
+
+func (s *jetstreamStore) CreateInvite(handle, role string, ttlSecs int) (Invite, error) {
+	if handle == "" {
+		return Invite{}, fmt.Errorf("membership: CreateInvite: handle required")
+	}
+	role, err := validateInviteRole(role)
+	if err != nil {
+		return Invite{}, err
+	}
+	token, err := newInviteToken()
+	if err != nil {
+		return Invite{}, err
+	}
+	now := time.Now().UTC()
+	inv := Invite{
+		Token:     token,
+		Handle:    handle,
+		Role:      role,
+		ExpiresAt: now.Add(inviteTTL(ttlSecs)).Format(time.RFC3339Nano),
+		Used:      false,
+		CreatedAt: now.Format(time.RFC3339Nano),
+	}
+	b, err := json.Marshal(inv)
+	if err != nil {
+		return Invite{}, fmt.Errorf("membership: marshal invite: %w", err)
+	}
+	ctx, cancel := s.ctx()
+	defer cancel()
+	// Create (not Put) so a token collision is rejected rather than silently
+	// overwriting a live invite — a 32-byte random collision is astronomically
+	// unlikely, but Create makes the single-use guarantee unconditional.
+	if _, err := s.invites.Create(ctx, token, b); err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return Invite{}, fmt.Errorf("membership: create invite: token collision")
+		}
+		return Invite{}, fmt.Errorf("membership: create invite: %w", err)
+	}
+	return inv, nil
+}
+
+func (s *jetstreamStore) GetInvite(token string) (Invite, error) {
+	ctx, cancel := s.ctx()
+	defer cancel()
+	e, err := s.invites.Get(ctx, token)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return Invite{}, fmt.Errorf("membership: get invite %q: %w", token, ErrNotFound)
+		}
+		return Invite{}, fmt.Errorf("membership: get invite %q: %w", token, err)
+	}
+	var inv Invite
+	if err := json.Unmarshal(e.Value(), &inv); err != nil {
+		return Invite{}, fmt.Errorf("membership: unmarshal invite: %w", err)
+	}
+	return inv, nil
+}
+
+func (s *jetstreamStore) ListInvites() ([]Invite, error) {
+	ctx, cancel := s.ctx()
+	w, err := s.invites.WatchAll(ctx, jetstream.IgnoreDeletes())
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("membership: list invites: %w", err)
+	}
+	defer cancel()
+	defer w.Stop()
+	var out []Invite
+	for {
+		select {
+		case e := <-w.Updates():
+			if e == nil {
+				sort.Slice(out, func(i, j int) bool {
+					if out[i].CreatedAt != out[j].CreatedAt {
+						return out[i].CreatedAt > out[j].CreatedAt // newest first
+					}
+					return out[i].Token < out[j].Token
+				})
+				return out, nil
+			}
+			var inv Invite
+			if err := json.Unmarshal(e.Value(), &inv); err != nil {
+				return nil, fmt.Errorf("membership: unmarshal invite: %w", err)
+			}
+			out = append(out, inv)
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+}
+
+// ConsumeInvite spends a KV invite and registers the presented signing key. With
+// no multi-key transaction, single-use is enforced by a compare-and-swap on the
+// invite: the token is marked used via Update against the revision read by Get,
+// so only ONE concurrent consumer can win the swap; the loser sees a revision
+// mismatch and is rejected as used. The user is registered AFTER the successful
+// swap. Burn-on-claim: if the signing key is already registered the swap has
+// already spent the token and we surface ErrUserExists — the SQLite store commits
+// the same way, so both backends behave identically.
+func (s *jetstreamStore) ConsumeInvite(token, signPub, kexPub string) error {
+	signPub = normalizeSignPub(signPub)
+	kexPub = normalizeSignPub(kexPub)
+	if signPub == "" {
+		return fmt.Errorf("membership: ConsumeInvite: sign_pub required")
+	}
+
+	ctx, cancel := s.ctx()
+	defer cancel()
+	e, err := s.invites.Get(ctx, token)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return fmt.Errorf("membership: consume invite %q: %w", token, ErrNotFound)
+		}
+		return fmt.Errorf("membership: consume invite %q: %w", token, err)
+	}
+	var inv Invite
+	if err := json.Unmarshal(e.Value(), &inv); err != nil {
+		return fmt.Errorf("membership: unmarshal invite: %w", err)
+	}
+	if inv.Used {
+		return fmt.Errorf("membership: consume invite %q: %w", token, ErrInviteUsed)
+	}
+	if inviteIsExpired(inv.ExpiresAt) {
+		return fmt.Errorf("membership: consume invite %q: %w", token, ErrInviteExpired)
+	}
+
+	inv.Used = true
+	inv.UsedAt = nowRFC3339()
+	inv.UsedSignPub = signPub
+	inv.UsedKexPub = kexPub
+	b, err := json.Marshal(inv)
+	if err != nil {
+		return fmt.Errorf("membership: marshal invite: %w", err)
+	}
+	// CAS: Update only succeeds if the invite is still at the revision we read, so
+	// a racing consumer that already flipped it loses here. A failed swap is
+	// conservatively treated as "already used" (the common cause); the caller can
+	// re-read to learn the precise state.
+	if _, err := s.invites.Update(ctx, token, b, e.Revision()); err != nil {
+		return fmt.Errorf("membership: consume invite %q: %w", token, ErrInviteUsed)
+	}
+
+	// Token is now spent. Register the user with the invite-fixed handle and role.
+	if err := s.AddUser(signPub, inv.Handle, inv.Role); err != nil {
+		if errors.Is(err, ErrUserExists) {
+			return ErrUserExists
+		}
+		return fmt.Errorf("membership: consume invite %q: register user: %w", token, err)
+	}
+	return nil
+}
+
+func (s *jetstreamStore) CancelInvite(token string) error {
+	ctx, cancel := s.ctx()
+	defer cancel()
+	if _, err := s.invites.Get(ctx, token); err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return fmt.Errorf("membership: cancel invite %q: %w", token, ErrNotFound)
+		}
+		return fmt.Errorf("membership: cancel invite %q: %w", token, err)
+	}
+	if err := s.invites.Delete(ctx, token); err != nil {
+		return fmt.Errorf("membership: cancel invite %q: %w", token, err)
+	}
+	return nil
+}
+
 // ---- snapshot import / export (issue 0003c migration) ---------------------

 // importSnapshot writes a full Snapshot into the KV buckets, preserving each