feat: implement freemium restrictions

go/cmd/orion/main.go

@@ -149,6 +149,9 @@ func main() {
 	mux.Handle("POST /networks/{id}/billing/checkout-session", withAuth(h.CreateCheckoutSession))
 	mux.Handle("POST /networks/{id}/billing/portal-session", withAuth(h.CreatePortalSession))
 
+	// Freemium usage (any network member)
+	mux.Handle("GET /networks/{id}/usage", withAuth(h.GetNetworkUsage))
+
 	// Network Invitations
 	mux.Handle("GET /networks/{id}/invitations", withAuth(h.ListInvitationsForNetwork))
 	mux.Handle("DELETE /networks/{id}/invitations", withAuth(h.RevokeInvitation))

go/cmd/particleprocessorworker/main.go

@@ -6,8 +6,10 @@ import (
 	"log"
 	"log/slog"
 	"os"
+	"strings"
 	"time"
 
+	"github.com/flowy-live/llink/internal/billing"
 	"github.com/flowy-live/llink/internal/db"
 	"github.com/flowy-live/llink/internal/depot"
 	"github.com/flowy-live/llink/internal/particle"
@@ -43,6 +45,7 @@ func main() {
 	db.Init()
 	defer db.Cleanup()
 	processingRepo := particle.NewProcessingRepository(db.Pool())
+	billingSvc := billing.NewServiceForWorker(db.Pool())
 
 	storageClient, err := storage.NewClient(ctx)
 	if err != nil {
@@ -104,6 +107,7 @@ func main() {
 
 			updateParentLastChildCreatedAt(ctx, change.Doc)
 			transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc)
+			recordFreemiumUsage(ctx, billingSvc, change.Doc)
 
 			if err := processingRepo.MarkProcessed(ctx, particleID); err != nil {
 				slog.Error("failed to mark particle as processed", "particleID", particleID, "error", err)
@@ -192,6 +196,59 @@ func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTr
 	}
 }
 
+// recordFreemiumUsage bumps the network's daily message counter for non-container
+// particles. Idempotent via the surrounding processed_particles guard: the worker
+// only reaches this path on first-seen particles, so a crash/restart won't
+// double-count.
+func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *firestore.DocumentSnapshot) {
+	rawType, err := doc.DataAt("type")
+	if err != nil {
+		slog.Error("failed to read particle type", "error", err, "particleID", doc.Ref.ID)
+		return
+	}
+	typeStr, ok := rawType.(string)
+	if !ok {
+		slog.Error("particle type is not a string", "particleID", doc.Ref.ID, "type", rawType)
+		return
+	}
+	particleType, err := particle.ParseParticleType(typeStr)
+	if err != nil {
+		slog.Error("invalid particle type", "error", err, "particleID", doc.Ref.ID)
+		return
+	}
+	// Containers (stream/folder) don't count as "messages" for the daily cap.
+	if particleType == particle.TypeStream || particleType == particle.TypeFolder {
+		return
+	}
+
+	networkID, err := networkIDFromParticlePath(doc.Ref.Path)
+	if err != nil {
+		slog.Error("failed to derive network id", "error", err, "path", doc.Ref.Path)
+		return
+	}
+
+	if err := billingSvc.IncrementDailyUsage(ctx, networkID, doc.CreateTime); err != nil {
+		slog.Error("failed to increment daily usage", "error", err, "networkID", networkID, "particleID", doc.Ref.ID)
+	}
+}
+
+// networkIDFromParticlePath extracts the network id from a Firestore particle
+// document path. Particles live at `networks/{network_id}/children/.../children/{id}`
+// at arbitrary nesting depth, so the network id is always the second segment
+// of the full doc path (which itself is rooted under the Firestore db path:
+// `projects/.../documents/networks/{network_id}/...`).
+func networkIDFromParticlePath(path string) (string, error) {
+	// doc.Ref.Path is the full resource path; find the "networks" collection
+	// and return the next segment.
+	segments := strings.Split(path, "/")
+	for i, seg := range segments {
+		if seg == "networks" && i+1 < len(segments) {
+			return segments[i+1], nil
+		}
+	}
+	return "", fmt.Errorf("no networks segment in path: %s", path)
+}
+
 // updateParentLastChildCreatedAt updates the parent stream's last_child_created_at
 // to the child's actual created_at timestamp, so it stays directly comparable with
 // playback markers (which also store child created_at values).

go/internal/billing/noop.go

@@ -3,6 +3,7 @@ package billing
 import (
 	"context"
 	"errors"
+	"time"
 )
 
 // Noop returns a billing service for binaries that depend on network.Service
@@ -24,3 +25,9 @@ func (noopService) CreatePortalSession(context.Context, string) (string, error)
 }
 func (noopService) SyncSeats(context.Context, string, int) error        { return nil }
 func (noopService) HandleWebhook(context.Context, []byte, string) error { return errNoopBilling }
+func (noopService) GetUsage(context.Context, string) (*Usage, error) {
+	return nil, errNoopBilling
+}
+func (noopService) IncrementDailyUsage(context.Context, string, time.Time) error {
+	return errNoopBilling
+}

go/internal/billing/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/stripe/stripe-go/v85"
@@ -22,6 +23,14 @@ type Service interface {
 	// No-op if the network has no active subscription.
 	SyncSeats(ctx context.Context, networkID string, seats int) error
 	HandleWebhook(ctx context.Context, payload []byte, signature string) error
+
+	// GetUsage reports today's freemium quota state for a network.
+	// Pro networks get Limit=nil (unlimited); free networks get Limit=&FreemiumDailyLimit.
+	GetUsage(ctx context.Context, networkID string) (*Usage, error)
+	// IncrementDailyUsage is called by the particle processor worker for each
+	// qualifying particle (non-container). Idempotency is the caller's concern
+	// — the worker guards this via processed_particles.
+	IncrementDailyUsage(ctx context.Context, networkID string, at time.Time) error
 }
 
 var (
@@ -32,6 +41,7 @@ var (
 type serviceImpl struct {
 	cfg               Config
 	repo              repository
+	usageRepo         usageRepository
 	priceMonthlyCents int64
 	priceAnnualCents  int64
 }
@@ -58,11 +68,22 @@ func NewService(ctx context.Context, pool *pgxpool.Pool, cfg Config) (Service, e
 	return &serviceImpl{
 		cfg:               cfg,
 		repo:              newRepository(pool),
+		usageRepo:         newUsageRepository(pool),
 		priceMonthlyCents: monthly.UnitAmount,
 		priceAnnualCents:  annual.UnitAmount,
 	}, nil
 }
 
+// NewServiceForWorker builds a minimal billing Service suitable for the
+// particle processor worker: only the usage-tracking path is exercised, so
+// we skip Stripe client setup (no API key required).
+func NewServiceForWorker(pool *pgxpool.Pool) Service {
+	return &serviceImpl{
+		usageRepo: newUsageRepository(pool),
+		repo:      newRepository(pool),
+	}
+}
+
 func (s *serviceImpl) GetStatus(ctx context.Context, networkID string) (*Status, error) {
 	sub, err := s.repo.getSubscriptionByNetworkID(ctx, networkID)
 	if err != nil && !errors.Is(err, errNotFound) {
@@ -208,6 +229,59 @@ func (s *serviceImpl) SyncSeats(ctx context.Context, networkID string, seats int
 	return nil
 }
 
+func (s *serviceImpl) IncrementDailyUsage(ctx context.Context, networkID string, at time.Time) error {
+	return s.usageRepo.incrementDaily(ctx, networkID, at)
+}
+
+func (s *serviceImpl) GetUsage(ctx context.Context, networkID string) (*Usage, error) {
+	now := time.Now()
+	used, err := s.usageRepo.getDaily(ctx, networkID, now)
+	if err != nil {
+		return nil, fmt.Errorf("read daily usage: %w", err)
+	}
+
+	plan, err := s.resolvePlan(ctx, networkID)
+	if err != nil {
+		return nil, err
+	}
+
+	u := &Usage{
+		Plan:    plan,
+		Used:    used,
+		ResetAt: nextUTCMidnight(now),
+	}
+	if plan == PlanFree {
+		limit := FreemiumDailyLimit
+		u.Limit = &limit
+	}
+	return u, nil
+}
+
+// resolvePlan is a lightweight read: it infers free/pro from the local
+// subscription row without hitting Stripe, so it is safe to call from the
+// particle processor worker (no Stripe client required).
+func (s *serviceImpl) resolvePlan(ctx context.Context, networkID string) (Plan, error) {
+	sub, err := s.repo.getSubscriptionByNetworkID(ctx, networkID)
+	if errors.Is(err, errNotFound) {
+		return PlanFree, nil
+	}
+	if err != nil {
+		return "", err
+	}
+	switch stripe.SubscriptionStatus(sub.Status) {
+	case stripe.SubscriptionStatusActive,
+		stripe.SubscriptionStatusTrialing,
+		stripe.SubscriptionStatusPastDue:
+		return PlanPro, nil
+	}
+	return PlanFree, nil
+}
+
+func nextUTCMidnight(now time.Time) time.Time {
+	utc := now.UTC()
+	return time.Date(utc.Year(), utc.Month(), utc.Day(), 0, 0, 0, 0, time.UTC).Add(24 * time.Hour)
+}
+
 func (s *serviceImpl) ensureStripeCustomer(ctx context.Context, p CheckoutParams) (string, error) {
 	existing, err := s.repo.getStripeCustomerID(ctx, p.NetworkID)
 	if err == nil {

go/internal/billing/usage_models.go

@@ -0,0 +1,15 @@
+package billing
+
+import "time"
+
+// FreemiumDailyLimit is the per-network daily cap on non-container
+// particles for networks on the free plan.
+const FreemiumDailyLimit = 50
+
+// Usage describes a network's current freemium quota state for today.
+type Usage struct {
+	Plan    Plan      `json:"plan"`
+	Used    int       `json:"used"`
+	Limit   *int      `json:"limit"` // nil = unlimited (pro)
+	ResetAt time.Time `json:"reset_at"`
+}

go/internal/billing/usage_repository.go

@@ -0,0 +1,49 @@
+package billing
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type usageRepository interface {
+	incrementDaily(ctx context.Context, networkID string, at time.Time) error
+	getDaily(ctx context.Context, networkID string, at time.Time) (int, error)
+}
+
+type usageRepositoryImpl struct {
+	pool *pgxpool.Pool
+}
+
+func newUsageRepository(pool *pgxpool.Pool) usageRepository {
+	return &usageRepositoryImpl{pool: pool}
+}
+
+func (r *usageRepositoryImpl) incrementDaily(ctx context.Context, networkID string, at time.Time) error {
+	_, err := r.pool.Exec(ctx, `
+		INSERT INTO network_message_usage (network_id, usage_date, message_count, updated_at)
+		VALUES ($1, ($2 AT TIME ZONE 'UTC')::date, 1, NOW())
+		ON CONFLICT (network_id, usage_date) DO UPDATE
+		SET message_count = network_message_usage.message_count + 1,
+		    updated_at    = NOW()
+	`, networkID, at)
+	return err
+}
+
+func (r *usageRepositoryImpl) getDaily(ctx context.Context, networkID string, at time.Time) (int, error) {
+	var count int
+	err := r.pool.QueryRow(ctx, `
+		SELECT message_count FROM network_message_usage
+		WHERE network_id = $1 AND usage_date = ($2 AT TIME ZONE 'UTC')::date
+	`, networkID, at).Scan(&count)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+	return count, nil
+}

go/internal/handler/billing.go

@@ -25,6 +25,46 @@ type PortalSessionResponse struct {
 	URL string `json:"url"`
 }
 
+// GetNetworkUsage returns the freemium quota state for the authenticated
+// caller's current network: how many messages they've used today, the daily
+// limit (null for pro), and when the counter resets.
+//
+// Authorization: any network member may read (not admin-only) since the UI
+// surfaces this to every sender.
+func (h *Handler) GetNetworkUsage(w http.ResponseWriter, r *http.Request) {
+	humanId, ok := middleware.HumanIdFromContext(r.Context())
+	if !ok {
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+
+	networkID := r.PathValue("id")
+	if networkID == "" {
+		http.Error(w, "network id is required", http.StatusBadRequest)
+		return
+	}
+
+	isMember, err := h.networkSvc.IsMember(r.Context(), networkID, humanId)
+	if err != nil {
+		slog.Error("failed to check network membership", "error", err, "network_id", networkID)
+		http.Error(w, "internal server error", http.StatusInternalServerError)
+		return
+	}
+	if !isMember {
+		http.Error(w, "forbidden", http.StatusForbidden)
+		return
+	}
+
+	usage, err := h.billingSvc.GetUsage(r.Context(), networkID)
+	if err != nil {
+		slog.Error("failed to get network usage", "error", err, "network_id", networkID)
+		http.Error(w, "internal server error", http.StatusInternalServerError)
+		return
+	}
+
+	writeJSON(w, usage)
+}
+
 func (h *Handler) GetNetworkBilling(w http.ResponseWriter, r *http.Request) {
 	net, _, ok := h.loadNetworkForAdmin(w, r)
 	if !ok {

go/migrations/000014_network_message_usage.down.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+DROP TABLE IF EXISTS network_message_usage;
+
+COMMIT;

go/migrations/000014_network_message_usage.up.sql

@@ -0,0 +1,11 @@
+BEGIN;
+
+CREATE TABLE network_message_usage (
+  network_id    TEXT        NOT NULL REFERENCES networks(id) ON DELETE CASCADE,
+  usage_date    DATE        NOT NULL,
+  message_count INTEGER     NOT NULL DEFAULT 0,
+  updated_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (network_id, usage_date)
+);
+
+COMMIT;