AppsGanin
diff --git a/‎cmd/rospanel/cli.go‎
Lines changed: 6 additions & 0 deletions b/‎cmd/rospanel/cli.go‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎cmd/rospanel/service.go‎
Lines changed: 36 additions & 4 deletions b/‎cmd/rospanel/service.go‎
Lines changed: 36 additions & 4 deletions
diff --git a/‎internal/core/manager.go‎
Lines changed: 86 additions & 7 deletions b/‎internal/core/manager.go‎
Lines changed: 86 additions & 7 deletions
diff --git a/‎internal/core/manager_access_test.go‎
Lines changed: 105 additions & 0 deletions b/‎internal/core/manager_access_test.go‎
Lines changed: 105 additions & 0 deletions
@@ -253,6 +253,12 @@ func runInstall() {
 		"ExecStart=" + installBinPath + "\n" +
 		"Restart=always\n" +
 		"RestartSec=3\n" +
+		// Signal only the panel, not everything in the cgroup. The default
+		// (control-group) SIGTERMs the Xray child too, so it would die on its own
+		// before the panel could mark the stop intentional — reported as a crash on
+		// every restart. The panel stops Xray itself; KillMode=mixed still SIGKILLs
+		// the whole group if the timeout expires, so nothing can be left behind.
+		"KillMode=mixed\n" +
 		// The panel still runs as root (it execs Xray, runs iptables/nft for the
 		// brute-guard + Hysteria port-hopping, writes net.* sysctls for BBR, and
 		// self-updates its own binary in /usr/local/bin). Rather than drop the user,
 
@@ -203,6 +203,9 @@ func runServer(dataDir string) {
 	go tlsLoop(mgr)
 	// Periodic traffic accounting + quota/expiry enforcement.
 	go statsPollLoop(mgr)
+	// Writes the buffered access-log sightings. RecordAccess only buffers, so this
+	// is what actually persists who connected from where.
+	go accessFlushLoop(mgr)
 	// Payment polling fallback: reconciles pending provider orders in case a webhook
 	// was missed. Idles cheaply when there are no pending orders.
 	go paymentPollLoop(mgr)
@@ -217,6 +220,13 @@ func runServer(dataDir string) {
 	// Telegram user bot: public self-service for VPN clients (registration,
 	// subscription, stats). Idles until enabled with its own token in Settings.
 	go telegram.NewUser(mgr, st).Run(context.Background())
+	// Telegram support bot: relays messages between a user's private chat and a
+	// per-user topic in the operator's forum supergroup. Idles until enabled with its
+	// own token and a group in Settings → Telegram.
+	go telegram.NewSupport(mgr, st).Run(context.Background())
+	// Broadcast delivery. Polls the store rather than holding a queue, so a restart
+	// mid-run resumes from the remaining recipients instead of losing or repeating.
+	go telegram.NewBroadcast(st, dataDir).Run(context.Background())
 
 	handler, err := server.New(mgr, secret, set.DecoyTemplate, dataDir)
 	if err != nil {
@@ -265,10 +275,16 @@ func runServer(dataDir string) {
 	<-stop
 	log.Print("shutting down")
 
+	// Stop Xray FIRST. Draining HTTP can take the full timeout below, and until the
+	// supervisor is marked closed an Xray exit still reads as a crash — which is
+	// exactly what happens under systemd's default KillMode, where Xray receives its
+	// own SIGTERM at the same moment we do. Marking the shutdown before we wait on
+	// anything is what keeps an ordinary restart from paging the operator.
+	sup.Stop()
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	_ = httpSrv.Shutdown(ctx)
-	sup.Stop()
 }
 
 // bootstrapTLS configures host/SNI and resolves a cert via ACME, falling back to
@@ -344,6 +360,21 @@ func statsPollLoop(mgr *core.Manager) {
 	}
 }
 
+// accessFlushInterval bounds how long a connection sighting sits in memory before
+// it is written, and so how quickly a newly-appeared device can trip the device
+// cap. Short enough to stay prompt, long enough that a busy server folds many
+// sightings into each commit.
+const accessFlushInterval = 5 * time.Second
+
+// accessFlushLoop persists buffered access-log sightings.
+func accessFlushLoop(mgr *core.Manager) {
+	t := time.NewTicker(accessFlushInterval)
+	defer t.Stop()
+	for range t.C {
+		safeTick("access flush", mgr.FlushAccess)
+	}
+}
+
 // paymentPollLoop reconciles pending provider orders (webhook fallback) every 25s.
 func paymentPollLoop(mgr *core.Manager) {
 	t := time.NewTicker(25 * time.Second)
@@ -353,14 +384,15 @@ func paymentPollLoop(mgr *core.Manager) {
 	}
 }
 
-// retentionLoop drops audit rows and stale connection rows past their retention
-// windows. Both cutoffs move by the day, so a slow cadence is plenty — this only
-// keeps the tables from growing forever.
+// retentionLoop drops audit rows, stale connection rows and old traffic history
+// past their retention windows. Every cutoff moves by the day, so a slow cadence is
+// plenty — this only keeps the tables from growing forever.
 func retentionLoop(mgr *core.Manager) {
 	sweep := func() {
 		mgr.PurgeOldEvents()
 		mgr.PurgeOldAdminAudit()
 		mgr.PurgeOldConnections()
+		mgr.PurgeOldTraffic()   // per-day traffic history past a year
 		mgr.PurgeExpiredUsers() // no-op unless the operator set a grace period
 		mgr.PurgeDeletedNodes() // reclaim node tombstones past their grace window
 	}
 
@@ -38,6 +38,10 @@ const reconcileDebounce = 800 * time.Millisecond
 const (
 	accLastMax = 4096
 	accLastTTL = int64(time.Hour / time.Second)
+	// accPendingMax bounds the unflushed sighting buffer. Sized above accLastMax so
+	// the throttle, not this cap, is what normally limits it — this only catches the
+	// pathological case where flushes keep failing and the buffer stops draining.
+	accPendingMax = 8192
 )
 
 // Manager is the application service layer.
@@ -53,6 +57,10 @@ type Manager struct {
 
 	accMu   sync.Mutex
 	accLast map[string]int64 // throttle key "uN|ip" → last recorded unix
+	// accPending buffers sightings between flushes, so the access-log reader never
+	// touches the database on the hot path. Bounded by the throttle above: one entry
+	// per user+IP per flush interval, not per log line.
+	accPending map[accPendingKey]store.ConnectionHit
 
 	// applyMu serializes config application (Reconcile + the live user-sync) so a
 	// direct Reconcile (e.g. from tlsLoop on cert renewal) can't interleave with the
@@ -81,8 +89,11 @@ type Manager struct {
 
 	// notifyThrottle bounds the rate of repeatable system alerts (Xray crash loop,
 	// cert renewal errors) so a stuck condition can't flood the admin chats.
-	throttleMu        sync.Mutex
-	lastCrashNotify   time.Time
+	throttleMu      sync.Mutex
+	lastCrashNotify time.Time
+	// crashAlerted records that admins were actually told about the current outage,
+	// so the all-clear is only sent for an alarm they saw.
+	crashAlerted      bool
 	lastCertErrNotify time.Time
 
 	// applyPlanMu serializes ApplyPlanToUser so the read-modify-write of expire_at
@@ -176,6 +187,7 @@ func New(st *store.Store, sup *xray.Supervisor, opts xray.Options, tls TLSPaths,
 		tls:              tls,
 		reconcileCh:      make(chan struct{}, 1),
 		accLast:          make(map[string]int64),
+		accPending:       make(map[accPendingKey]store.ConnectionHit),
 		applied:          make(map[int64]struct{}),
 		tz:               time.Local,
 		guard:            newBruteGuard(),
@@ -205,7 +217,8 @@ func New(st *store.Store, sup *xray.Supervisor, opts xray.Options, tls TLSPaths,
 			go func() { _ = m.syncOpera(true, set.OperaCountryOr(), set.OperaPortOr()) }()
 		}
 	}
-	m.sup.SetOnCrash(m.onXrayCrash) // alert admins when Xray exits unexpectedly
+	m.sup.SetOnCrash(m.onXrayCrash)     // alert admins when Xray exits unexpectedly
+	m.sup.SetOnRecover(m.onXrayRecover) // ...and tell them when it is back
 	go m.reconcileLoop()
 	go m.proxyLoop()
 	go m.geoLoop()    // auto-refresh geo databases on the operator's cadence
@@ -248,8 +261,18 @@ func (m *Manager) loc() *time.Location {
 // Location exposes the operator timezone for handlers that compute date ranges.
 func (m *Manager) Location() *time.Location { return m.loc() }
 
+// accPendingKey identifies one buffered sighting.
+type accPendingKey struct {
+	userID int64
+	ip     string
+}
+
 // RecordAccess notes a connection from an Xray access-log line (email "uN" +
-// source IP). Throttled to one DB write per user+IP per 10s to absorb bursts.
+// source IP). Throttled to one recorded sighting per user+IP per 10s to absorb
+// bursts, then buffered — FlushAccess writes them.
+//
+// This is called from the access-log reader for every line Xray emits, so it does
+// no I/O at all: it takes a lock, updates two maps, and returns.
 func (m *Manager) RecordAccess(email, ip string) {
 	if !strings.HasPrefix(email, "u") {
 		return
@@ -261,8 +284,8 @@ func (m *Manager) RecordAccess(email, ip string) {
 	now := time.Now().Unix()
 	key := email + "|" + ip
 	m.accMu.Lock()
+	defer m.accMu.Unlock()
 	if now-m.accLast[key] < 10 {
-		m.accMu.Unlock()
 		return
 	}
 	m.accLast[key] = now
@@ -273,14 +296,70 @@ func (m *Manager) RecordAccess(email, ip string) {
 			}
 		}
 	}
+	pk := accPendingKey{userID: id, ip: ip}
+	h, buffered := m.accPending[pk]
+	// Bound the buffer. It normally drains every few seconds, but a persistent write
+	// failure (a full disk, say) makes FlushAccess requeue instead — and the throttle
+	// above stops protecting us as soon as accLast evicts a key, since that reopens
+	// the pair for buffering. Dropping the newest sighting for a pair we are not
+	// already tracking costs a last_seen update; growing without limit costs the
+	// process.
+	if !buffered && len(m.accPending) >= accPendingMax {
+		return
+	}
+	h.UserID, h.IP, h.Hits = id, ip, h.Hits+1
+	if now > h.SeenAt {
+		h.SeenAt = now
+	}
+	m.accPending[pk] = h
+}
+
+// FlushAccess writes the buffered access sightings in one transaction and, if the
+// new devices changed who should be online, syncs Xray.
+//
+// The device-cap re-check used to run per sighting — a full WorkingUsers query for
+// every user+IP every 10s. It belongs here: it only has to happen when something
+// was actually recorded, and once per batch answers the same question.
+func (m *Manager) FlushAccess() {
+	m.accMu.Lock()
+	if len(m.accPending) == 0 {
+		m.accMu.Unlock()
+		return
+	}
+	hits := make([]store.ConnectionHit, 0, len(m.accPending))
+	for _, h := range m.accPending {
+		hits = append(hits, h)
+	}
+	clear(m.accPending)
 	m.accMu.Unlock()
-	if err := m.store.AddConnection(id, ip, now); err != nil {
+
+	if err := m.store.AddConnections(hits); err != nil {
+		// Put them back rather than drop them: the buffer was already drained, so
+		// returning here would silently lose the sightings, and stale last_seen /
+		// undercounted devices feed straight into the device cap. Merging (rather than
+		// overwriting) keeps whatever arrived while the write was in flight.
+		m.accMu.Lock()
+		for _, h := range hits {
+			pk := accPendingKey{userID: h.UserID, ip: h.IP}
+			cur, buffered := m.accPending[pk]
+			if !buffered && len(m.accPending) >= accPendingMax {
+				continue // same bound as RecordAccess: shed rather than grow forever
+			}
+			cur.UserID, cur.IP = h.UserID, h.IP
+			cur.Hits += h.Hits
+			if h.SeenAt > cur.SeenAt {
+				cur.SeenAt = h.SeenAt
+			}
+			m.accPending[pk] = cur
+		}
+		m.accMu.Unlock()
+		logErr("access: flush failed, sightings requeued", "sightings", len(hits), "err", err)
 		return
 	}
 	// A new device (source IP) may push the user over their device cap — re-check
 	// the working set and sync promptly so the over-limit user drops out, instead
 	// of waiting for the next periodic reconcile.
-	if working, err := m.store.WorkingUsers(now); err == nil && m.workingChanged(working) {
+	if working, err := m.store.WorkingUsers(time.Now().Unix()); err == nil && m.workingChanged(working) {
 		m.TriggerUserSync()
 	}
 }
 
@@ -0,0 +1,105 @@
+package core
+
+import (
+	"fmt"
+	"path/filepath"
+	"testing"
+
+	"github.com/AppsGanin/rospanel/internal/store"
+)
+
+func accessTestManager(t *testing.T) (*Manager, *store.Store) {
+	t.Helper()
+	st, err := store.Open(filepath.Join(t.TempDir(), "acc.db"))
+	if err != nil {
+		t.Fatalf("open store: %v", err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return &Manager{
+		store:      st,
+		accLast:    make(map[string]int64),
+		accPending: make(map[accPendingKey]store.ConnectionHit),
+	}, st
+}
+
+// TestRecordAccessDoesNoIO pins the hot path. RecordAccess runs for every line the
+// Xray access log emits; it must only touch memory, leaving the write to the
+// flusher. Before this, each admitted sighting did two statements plus a full
+// WorkingUsers query — the panel's single busiest write source.
+func TestRecordAccessDoesNoIO(t *testing.T) {
+	m, st := accessTestManager(t)
+	u, err := st.CreateUser("u1", "uuid-1", "pw", "tok", 0, 0, 0)
+	if err != nil {
+		t.Fatalf("create user: %v", err)
+	}
+	email := fmt.Sprintf("u%d", u.ID)
+
+	for range 50 {
+		m.RecordAccess(email, "1.1.1.1")
+		m.RecordAccess(email, "2.2.2.2")
+	}
+
+	conns, err := st.RecentConnections(u.ID, 10)
+	if err != nil {
+		t.Fatalf("connections: %v", err)
+	}
+	if len(conns) != 0 {
+		t.Fatalf("RecordAccess wrote %d rows before the flush — the hot path is still doing I/O", len(conns))
+	}
+	// The 10s throttle collapses the burst: two IPs, one buffered sighting each.
+	if got := len(m.accPending); got != 2 {
+		t.Fatalf("buffered %d sightings from 100 calls across 2 IPs, want 2", got)
+	}
+}
+
+// TestFlushAccessWritesBatch: the buffered sightings land, and flushing an empty
+// buffer is free.
+func TestFlushAccessWritesBatch(t *testing.T) {
+	m, st := accessTestManager(t)
+	u, err := st.CreateUser("u1", "uuid-1", "pw", "tok", 0, 0, 0)
+	if err != nil {
+		t.Fatalf("create user: %v", err)
+	}
+	email := fmt.Sprintf("u%d", u.ID)
+	for _, ip := range []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"} {
+		m.RecordAccess(email, ip)
+	}
+
+	m.FlushAccess()
+	conns, err := st.RecentConnections(u.ID, 10)
+	if err != nil {
+		t.Fatalf("connections: %v", err)
+	}
+	if len(conns) != 3 {
+		t.Fatalf("flushed %d connection rows, want 3", len(conns))
+	}
+	cur, err := st.GetUser(u.ID)
+	if err != nil {
+		t.Fatalf("get user: %v", err)
+	}
+	if cur.LastSeen == 0 {
+		t.Error("flush did not stamp last_seen")
+	}
+
+	// Buffer is drained, so a second flush writes nothing new.
+	if len(m.accPending) != 0 {
+		t.Fatalf("buffer still holds %d sightings after a flush", len(m.accPending))
+	}
+	m.FlushAccess()
+	conns, _ = st.RecentConnections(u.ID, 10)
+	if len(conns) != 3 {
+		t.Fatalf("second flush changed the row count to %d", len(conns))
+	}
+}
+
+// TestRecordAccessIgnoresJunk: the access log is parsed text, so non-user emails
+// must not create buffer entries.
+func TestRecordAccessIgnoresJunk(t *testing.T) {
+	m, _ := accessTestManager(t)
+	for _, email := range []string{"", "admin", "unotanumber", "12", "u"} {
+		m.RecordAccess(email, "1.1.1.1")
+	}
+	if len(m.accPending) != 0 {
+		t.Fatalf("buffered %d sightings from junk emails", len(m.accPending))
+	}
+}