subscriber

package
v4.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 12, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package subscriber is the kernel's single writer of shared subscriber state: entitlements, traffic counters, subscription credentials and balances in v2_user and v2_user_subscription_group (docs/architecture/subscriber-service.md). The kernel's own callers and the KernelSubscriber contract use the same functions, so both behave alike. Every retryable write carries a request id, applied once.

Index

Constants

View Source
const (
	BalanceAccount    = "balance"
	BalanceCommission = "commission_balance"
)

Balance kinds.

View Source
const ActivityWriteInterval = time.Minute

ActivityWriteInterval is the least time between two writes of one user's last-online time. A node reports a user's traffic and connections every minute or so; the stored time lags the report by less than this.

View Source
const ChangeRetention = 7 * 24 * time.Hour

ChangeRetention is how long the change log keeps rows; a consumer whose cursor is older lists again.

View Source
const RequestRetention = 90 * 24 * time.Hour

RequestRetention is how long the request ledger keeps a request id; a retry after that applies again.

Variables

View Source
var ErrInsufficientBalance = errors.New("insufficient balance")

ErrInsufficientBalance means a debit would take a balance below zero.

View Source
var ErrMembershipNotFound = errors.New("subscription group membership not found")

ErrMembershipNotFound means the subscriber does not hold the group.

View Source
var ErrNegativeTraffic = errors.New("traffic must not be negative")

ErrNegativeTraffic means a traffic entry would decrease a counter.

View Source
var ErrSubscriberNotFound = errors.New("subscriber not found")

ErrSubscriberNotFound means no v2_user row has the id.

View Source
var ErrSubscriptionGroupNotFound = errors.New("subscription group not found")

ErrSubscriptionGroupNotFound means no v2_subscription_group row has the id.

View Source
var PeriodMonths = map[string]int{
	"month": 1, "quarter": 3, "half_year": 6, "year": 12, "two_year": 24, "three_year": 36, "onetime": 1200,
}

PeriodMonths maps a v2 order period to months.

Functions

func Active

func Active(db *gorm.DB, now time.Time) *gorm.DB

Active restricts a v2_user query to the subscribers a node serves: not banned, not expired, and within their traffic (a zero transfer limit serves nothing), as node user lists always have.

func ChangesAfter

func ChangesAfter(db *gorm.DB, cursor uint64, limit int) (changes []model.SubscriberChange, resync bool, err error)

ChangesAfter returns up to limit changes after cursor. resync is true when rows after the cursor may have been pruned: the consumer must list again.

func Cursor

func Cursor(db *gorm.DB) (uint64, error)

Cursor returns the latest change cursor (0 when the log is empty).

func ForgetActivityTx added in v4.2.0

func ForgetActivityTx(tx *gorm.DB, userID uint) error

ForgetActivityTx drops a deleted user's last-online time in tx, so a later user that gets the same id (SQLite reuses the highest one) does not inherit it. A database without the table is left alone.

func LastOnline added in v4.2.0

func LastOnline(db *gorm.DB, userIDs []uint) (map[uint]int64, error)

LastOnline returns the last-online time (a Unix time) of each of the users that has one; a user never seen online is not in the map.

func PruneChanges

func PruneChanges(db *gorm.DB, now time.Time) (int64, error)

PruneChanges deletes change log rows older than the retention.

func PruneRequests

func PruneRequests(db *gorm.DB, now time.Time) (int64, error)

PruneRequests deletes request ledger rows older than RequestRetention.

func Record

func Record(tx *gorm.DB, requestID, method string, userID uint, result any, now time.Time) error

Record stores a request id with its result in tx.

func RecordChangesTx

func RecordChangesTx(tx *gorm.DB, userIDs []uint, deleted bool, now time.Time) error

RecordChangesTx appends change log rows for the subscribers in tx, in the given order.

func RecordNodeGroupChangeTx

func RecordNodeGroupChangeTx(tx *gorm.DB, previous, next *uint, now time.Time) error

RecordNodeGroupChangeTx appends a change for each active subscriber a node served or now serves when its group moves from previous to next (nil is every subscriber). The node's users are derived from its group, so a watcher of the node would otherwise keep the old group's users and never get the new group's until each of them changed: the changes make it re-read those subscribers on the node, adding the new group's and removing the old group's. Nothing is recorded when the group is the same.

func RecordOnline added in v4.2.0

func RecordOnline(db *gorm.DB, userIDs []uint, at time.Time)

RecordOnline notes that the users were seen online at at: their traffic was reported, or a node listed their connections. It is the one writer of v4_kernel_user_activity and is best effort: it never fails or delays the report it belongs to (call it after the report committed), and an error is logged and not retried before the interval passes. Each user is written at most once per ActivityWriteInterval, in multi-row upserts that only move a stored time forward and only for users that exist, so concurrent Control processes and late reports cannot move it back.

func Replay

func Replay(tx *gorm.DB, requestID string, result any) (bool, error)

Replay loads the recorded result of requestID into result and reports whether it was applied before.

func ResetActivityThrottleForTest added in v4.2.0

func ResetActivityThrottleForTest(db *gorm.DB)

ResetActivityThrottleForTest forgets which users were written on db, so a test can record the same user again within the interval.

func TouchesNodeFields

func TouchesNodeFields(updates map[string]any) bool

TouchesNodeFields reports whether a v2_user update changes a column that node user lists depend on.

Types

type BalanceResult

type BalanceResult struct {
	Applied bool  `json:"-"`
	Balance int64 `json:"balance"`
}

BalanceResult is the outcome of AdjustBalanceTx.

func AdjustBalanceTx

func AdjustBalanceTx(tx *gorm.DB, requestID string, userID uint, kind string, amount int64, now time.Time) (BalanceResult, error)

AdjustBalanceTx adds amount (cents, possibly negative) to a subscriber's account or commission balance in tx, once per request id.

type Entitlement

type Entitlement struct {
	// RequestID is unique per grant, e.g. "order:42"; empty means the
	// write is not retryable and is not recorded.
	RequestID string
	UserID    uint
	Plan      PlanSnapshot
	// Period extends the expiry from now, or with RenewSamePlan from the
	// current expiry of the same, unexpired plan. ExpiresAt sets it
	// exactly. With neither, the expiry is unchanged.
	Period                 *Period
	ExpiresAt              *int64
	RenewSamePlan          bool
	ResetTraffic           bool
	KeepSubscriptionGroups bool
}

Entitlement activates or renews a plan.

type EntitlementResult

type EntitlementResult struct {
	// Applied is false when the request id was applied before; ExpiresAt
	// then describes that first application.
	Applied   bool   `json:"-"`
	ExpiresAt *int64 `json:"expires_at,omitempty"`
}

EntitlementResult is the outcome of ApplyEntitlementTx.

func ApplyEntitlementTx

func ApplyEntitlementTx(tx *gorm.DB, e Entitlement, now time.Time) (EntitlementResult, error)

ApplyEntitlementTx applies an entitlement in tx, under the subscriber's row lock.

type GrantResult

type GrantResult struct {
	// Applied is false when the request id was applied before; Created
	// then describes that first application.
	Applied bool `json:"-"`
	Created bool `json:"created"`
}

GrantResult is the outcome of GrantSubscriptionGroupTx.

func GrantSubscriptionGroupTx

func GrantSubscriptionGroupTx(tx *gorm.DB, g GroupGrant, now time.Time) (GrantResult, error)

GrantSubscriptionGroupTx gives a subscriber a subscription group in tx, under the subscriber's row lock, once per request id: it creates the membership, or sets the given fields of an existing one. A subscriber or group that does not exist is an error, and nothing is recorded. A new membership, or a new expiry, records a change for an active subscriber.

type GroupGrant

type GroupGrant struct {
	// RequestID is unique per grant; empty means the write is not
	// retryable and is not recorded.
	RequestID string
	UserID    uint
	GroupID   uint
	// ExpiresAt, TransferBytes and NextRenewPrice are set when non-nil and
	// left unchanged otherwise, as given.
	ExpiresAt      *int64
	TransferBytes  *int64
	NextRenewPrice *int64
}

GroupGrant gives a subscriber one subscription group.

type MembershipResult

type MembershipResult struct {
	Applied bool `json:"-"`
	// Removed counts the memberships deleted.
	Removed int64 `json:"removed"`
}

MembershipResult is the outcome of RevokeSubscriptionGroupTx and RemoveSubscriptionGroupMembersTx.

func RemoveSubscriptionGroupMembersTx

func RemoveSubscriptionGroupMembersTx(tx *gorm.DB, requestID string, groupID uint, now time.Time) (MembershipResult, error)

RemoveSubscriptionGroupMembersTx takes a subscription group from every subscriber who holds it in tx, under their row locks (taken in id order), once per request id, whether or not the group still exists. It records a change for each active member.

func RevokeSubscriptionGroupTx

func RevokeSubscriptionGroupTx(tx *gorm.DB, requestID string, userID, groupID uint, now time.Time) (MembershipResult, error)

RevokeSubscriptionGroupTx takes a subscription group from a subscriber in tx, under the subscriber's row lock, once per request id. A subscriber, group or membership that does not exist is an error, and nothing is recorded. It records a change for an active subscriber.

type Period

type Period struct {
	Months int
	Days   int
}

Period is how long an entitlement lasts.

type PlanSnapshot

type PlanSnapshot struct {
	PlanID  uint
	GroupID uint
	// SubscriptionGroupIDs replace the subscriber's subscription groups,
	// unless the entitlement keeps them.
	SubscriptionGroupIDs []uint
	TransferBytes        int64
	// SpeedLimit and DeviceLimit are left unchanged when nil.
	SpeedLimit  *int64
	DeviceLimit *int
}

PlanSnapshot is what an entitlement grants.

func PlanSnapshotFromPlan

func PlanSnapshotFromPlan(plan model.Plan, groups []model.PlanSubscriptionGroup) PlanSnapshot

PlanSnapshotFromPlan builds a snapshot of a v2 plan and its subscription groups.

type ResetResult

type ResetResult struct {
	Applied bool  `json:"-"`
	Reset   int64 `json:"reset"`
}

ResetResult is the outcome of ResetTrafficTx.

func ResetTrafficTx

func ResetTrafficTx(tx *gorm.DB, requestID string, userIDs []uint, now time.Time) (ResetResult, error)

ResetTrafficTx zeroes subscribers' counters in tx, once per request id.

type TrafficEntry

type TrafficEntry struct {
	UserID   uint
	Upload   int64
	Download int64
}

TrafficEntry is traffic to add to one subscriber's counters, in bytes already multiplied by any rate.

type TrafficResult

type TrafficResult struct {
	Applied bool `json:"-"`
	// Exhausted lists subscribers who crossed their transfer limit.
	Exhausted []uint `json:"exhausted,omitempty"`
}

TrafficResult is the outcome of RecordTrafficTx.

func RecordTrafficTx

func RecordTrafficTx(tx *gorm.DB, batchID string, entries []TrafficEntry, now time.Time) (TrafficResult, error)

RecordTrafficTx adds traffic to subscribers' counters in tx, once per batch id (an empty batch id is not deduplicated: legacy node reports carry none). Entries for the same subscriber are summed, rows are updated in id order so concurrent reports cannot deadlock, and unknown subscribers are skipped, as before.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL