fenrir/internal/notifications/shared_linux.go

// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later

// Package notifications shares silent history and cue admission between readers.
// The first reader hosts an in-memory coordinator on a per-user abstract Unix
// socket. No notification text is written to disk and no service manager is used.
package notifications

import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"os"
	"slices"
	"strings"
	"sync"
	"syscall"
	"time"
	"unicode/utf8"
)

const Limit = 100
const SoundInterval = 30 * time.Second
const exchangeTimeout = 400 * time.Millisecond

type Entry struct {
	ID       string
	Text     string
	Received time.Time
}

type Snapshot struct {
	Entries   []Entry
	LastSound time.Time
}

type request struct {
	Version int
	Seed    Snapshot
	Claim   bool
	Member  string
	Leave   bool
}

type response struct {
	Version int
	State   Snapshot
	Allowed bool
}

type Client struct {
	UID       int
	id        string
	address   string
	ctx       context.Context
	cancel    context.CancelFunc
	mu        sync.Mutex
	state     Snapshot
	brokerMu  sync.Mutex
	broker    *broker
	wake      chan struct{}
	updates   chan struct{}
	sounds    chan func(context.Context)
	done      chan struct{}
	closeOnce sync.Once
}

// New starts only private IPC work. Namespace isolates native tests from live
// readers; production uses the empty namespace and the kernel account ID.
func New(ctx context.Context, uid int, namespace string) *Client {
	ctx, cancel := context.WithCancel(ctx)
	c := &Client{UID: uid, id: fmt.Sprintf("%x", random_id()), address: fmt.Sprintf("@fenrir-notifications.%d%s", uid, namespace), ctx: ctx, cancel: cancel,
		wake: make(chan struct{}, 1), updates: make(chan struct{}, 1), sounds: make(chan func(context.Context), 1), done: make(chan struct{})}
	go c.run()
	return c
}

func (c *Client) Updates() <-chan struct{} { return c.updates }

func (c *Client) Snapshot() Snapshot {
	c.mu.Lock()
	defer c.mu.Unlock()
	return Snapshot{Entries: slices.Clone(c.state.Entries), LastSound: c.state.LastSound}
}

func (c *Client) Publish(text string, received time.Time) {
	text = strings.TrimSpace(text)
	if text == "" {
		return
	}
	c.adopt(Snapshot{Entries: []Entry{{ID: fmt.Sprintf("%x", random_id()), Text: text, Received: received.UTC()}}})
	signal(c.wake)
}

func random_id() [16]byte {
	var id [16]byte
	// crypto/rand.Read cannot return an error with the Go version we require.
	_, _ = rand.Read(id[:])
	return id
}

func signal(channel chan struct{}) {
	select {
	case channel <- struct{}{}:
	default:
	}
}

func (c *Client) adopt(state Snapshot) {
	c.mu.Lock()
	before := c.state
	c.state = merge(c.state, state)
	changed := !slices.Equal(before.Entries, c.state.Entries)
	c.mu.Unlock()
	if changed {
		signal(c.updates)
	}
}

// Claim_sound fails closed if coordination is unavailable. Notification speech
// never depends on its success, and history synchronization cannot call speech.
func (c *Client) Claim_sound(ctx context.Context) bool {
	if last := c.Snapshot().LastSound; !last.IsZero() && time.Since(last) < SoundInterval {
		return false
	}
	result, err := c.exchange(ctx, true)
	return err == nil && result.Allowed
}

// Sound is for muted notifications: coalesce pending cue requests without
// running network or sound waits on the reader's input loop.
func (c *Client) Sound(play func(context.Context)) {
	select {
	case c.sounds <- play:
	default:
	}
}

func (c *Client) run() {
	defer close(c.done)
	tick := time.NewTicker(200 * time.Millisecond)
	defer tick.Stop()
	var previousError string
	for {
		_, err := c.exchange(c.ctx, false)
		if err != nil && c.ctx.Err() == nil && err.Error() != previousError {
			slog.Warn("notification history synchronization failed", "error", err, "at", time.Now())
			previousError = err.Error()
		} else if err == nil {
			previousError = ""
		}
		select {
		case <-c.ctx.Done():
			return
		case <-c.wake:
		case <-tick.C:
		case play := <-c.sounds:
			if c.Claim_sound(c.ctx) {
				play(c.ctx)
			}
		}
	}
}

func (c *Client) Close() {
	c.closeOnce.Do(func() {
		c.cancel()
		<-c.done
		// Flush the local outbox even when the reader's parent was canceled.
		_, _ = c.transact(context.Background(), false, true)
		c.brokerMu.Lock()
		if c.broker != nil {
			c.broker.close()
		}
		c.brokerMu.Unlock()
	})
}

func (c *Client) exchange(ctx context.Context, claim bool) (response, error) {
	return c.transact(ctx, claim, false)
}

func (c *Client) transact(ctx context.Context, claim, leaving bool) (response, error) {
	var result response
	ctx, cancel := context.WithTimeout(ctx, exchangeTimeout)
	defer cancel()
	if !leaving {
		stop := context.AfterFunc(c.ctx, cancel)
		defer stop()
	}
	if c.UID < 0 || os.Geteuid() != 0 && os.Geteuid() != c.UID {
		return result, fmt.Errorf("notification account is not permitted")
	}
	if err := ctx.Err(); err != nil {
		return result, err
	}
	dialer := net.Dialer{}
	connection, err := dialer.DialContext(ctx, "unix", c.address)
	if errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ENOENT) {
		if !leaving {
			if err := c.host(); err != nil && !errors.Is(err, syscall.EADDRINUSE) {
				return result, err
			}
			connection, err = dialer.DialContext(ctx, "unix", c.address)
		}
	}
	if err != nil {
		return result, err
	}
	defer connection.Close()
	stopConnection := context.AfterFunc(ctx, func() { connection.Close() })
	defer stopConnection()
	uid, err := peer_uid(connection.(*net.UnixConn))
	if err != nil || uid != c.UID && uid != 0 {
		return result, fmt.Errorf("notification coordinator belongs to another account")
	}
	deadline, _ := ctx.Deadline()
	_ = connection.SetDeadline(deadline)
	if err := json.NewEncoder(connection).Encode(request{Version: 1, Seed: c.Snapshot(), Claim: claim, Member: c.id, Leave: leaving}); err != nil {
		return result, err
	}
	if err := json.NewDecoder(io.LimitReader(connection, 8<<20)).Decode(&result); err != nil {
		return result, err
	}
	if result.Version != 1 || !valid(result.State) {
		return response{}, fmt.Errorf("invalid notification snapshot")
	}
	c.adopt(result.State)
	return result, nil
}

func (c *Client) host() error {
	c.brokerMu.Lock()
	defer c.brokerMu.Unlock()
	if err := c.ctx.Err(); err != nil {
		return err
	}
	listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: c.address, Net: "unix"})
	if err != nil {
		return err
	}
	c.broker = &broker{listener: listener, uid: c.UID, state: c.Snapshot(), done: make(chan struct{}), members: make(map[string]acknowledgement), changed: make(chan struct{}, 1)}
	go c.broker.serve()
	return nil
}

type acknowledgement struct {
	state Snapshot
	seen  time.Time
}

type broker struct {
	listener *net.UnixListener
	uid      int
	mu       sync.Mutex
	state    Snapshot
	workers  sync.WaitGroup
	done     chan struct{}
	members  map[string]acknowledgement
	changed  chan struct{}
}

func (b *broker) serve() {
	defer close(b.done)
	// Bound resource use independently of the reader's ordinary remote sockets.
	clients := make(chan struct{}, 16)
	for {
		connection, err := b.listener.AcceptUnix()
		if err != nil {
			return
		}
		select {
		case clients <- struct{}{}:
			b.workers.Go(func() {
				defer func() { connection.Close(); <-clients }()
				b.handle(connection)
			})
		default:
			connection.Close()
		}
	}
}

func (b *broker) handle(connection *net.UnixConn) {
	uid, err := peer_uid(connection)
	if err != nil || uid != 0 && uid != b.uid {
		return
	}
	_ = connection.SetDeadline(time.Now().Add(exchangeTimeout))
	var message request
	if json.NewDecoder(io.LimitReader(connection, 8<<20)).Decode(&message) != nil || message.Version != 1 || len(message.Member) != 32 || !valid(message.Seed) {
		return
	}
	b.mu.Lock()
	now := time.Now()
	for id, member := range b.members {
		if now.Sub(member.seen) > 5*time.Second {
			delete(b.members, id)
		}
	}
	// A crashed reader cannot deregister. Bound retained acknowledgement state
	// as well as concurrent connections; history itself is always capped at 100.
	if _, exists := b.members[message.Member]; !message.Leave && !exists && len(b.members) >= 128 {
		b.mu.Unlock()
		return
	}
	b.state = merge(b.state, message.Seed)
	if message.Leave {
		delete(b.members, message.Member)
	} else {
		b.members[message.Member] = acknowledgement{state: message.Seed, seen: now}
	}
	allowed := false
	if message.Claim && (b.state.LastSound.IsZero() || now.Sub(b.state.LastSound) >= SoundInterval) {
		b.state.LastSound, allowed = now, true
	}
	result := response{Version: 1, State: b.state, Allowed: allowed}
	b.mu.Unlock()
	signal(b.changed)
	_ = json.NewEncoder(connection).Encode(result)
}

func (b *broker) close() {
	// Give surviving replicas time to acknowledge the final snapshot before
	// releasing the coordinator's socket. This runs after releasing keyboards.
	deadline := time.NewTimer(3 * exchangeTimeout)
	defer deadline.Stop()
replicate:
	for {
		b.mu.Lock()
		replicated := true
		for _, member := range b.members {
			state := member.state
			merged := merge(state, b.state)
			if !slices.Equal(merged.Entries, state.Entries) || merged.LastSound.After(state.LastSound) {
				replicated = false
				break
			}
		}
		b.mu.Unlock()
		if replicated {
			break
		}
		select {
		case <-b.changed:
		case <-deadline.C:
			break replicate
		}
	}
	b.listener.Close()
	<-b.done
	b.workers.Wait()
}

func peer_uid(connection *net.UnixConn) (int, error) {
	raw, err := connection.SyscallConn()
	if err != nil {
		return -1, err
	}
	var credentials *syscall.Ucred
	var credentialError error
	err = raw.Control(func(fd uintptr) {
		credentials, credentialError = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED)
	})
	if err != nil {
		return -1, err
	}
	if credentialError != nil {
		return -1, credentialError
	}
	return int(credentials.Uid), nil
}

func valid(state Snapshot) bool {
	if len(state.Entries) > Limit || state.LastSound.After(time.Now().Add(time.Second)) {
		return false
	}
	seen := make(map[string]bool, len(state.Entries))
	for _, entry := range state.Entries {
		if len(entry.ID) != 32 || seen[entry.ID] || strings.TrimSpace(entry.Text) == "" || len(entry.Text) > 8129 || !utf8.ValidString(entry.Text) || entry.Received.IsZero() {
			return false
		}
		seen[entry.ID] = true
	}
	return true
}

func merge(left, right Snapshot) Snapshot {
	entries := make(map[string]Entry, len(left.Entries)+len(right.Entries))
	for _, entry := range right.Entries {
		entries[entry.ID] = entry
	}
	for _, entry := range left.Entries {
		entries[entry.ID] = entry
	}
	result := Snapshot{LastSound: left.LastSound}
	if right.LastSound.After(result.LastSound) {
		result.LastSound = right.LastSound
	}
	for _, entry := range entries {
		result.Entries = append(result.Entries, entry)
	}
	slices.SortFunc(result.Entries, func(a, b Entry) int {
		if order := a.Received.Compare(b.Received); order != 0 {
			return order
		}
		return strings.Compare(a.ID, b.ID)
	})
	if len(result.Entries) > Limit {
		result.Entries = result.Entries[len(result.Entries)-Limit:]
	}
	return result
}