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
- Variables
- func Active(db *gorm.DB, now time.Time) *gorm.DB
- func ChangesAfter(db *gorm.DB, cursor uint64, limit int) (changes []model.SubscriberChange, resync bool, err error)
- func Cursor(db *gorm.DB) (uint64, error)
- func ForgetActivityTx(tx *gorm.DB, userID uint) error
- func LastOnline(db *gorm.DB, userIDs []uint) (map[uint]int64, error)
- func PruneChanges(db *gorm.DB, now time.Time) (int64, error)
- func PruneRequests(db *gorm.DB, now time.Time) (int64, error)
- func Record(tx *gorm.DB, requestID, method string, userID uint, result any, now time.Time) error
- func RecordChangesTx(tx *gorm.DB, userIDs []uint, deleted bool, now time.Time) error
- func RecordNodeGroupChangeTx(tx *gorm.DB, previous, next *uint, now time.Time) error
- func RecordOnline(db *gorm.DB, userIDs []uint, at time.Time)
- func Replay(tx *gorm.DB, requestID string, result any) (bool, error)
- func ResetActivityThrottleForTest(db *gorm.DB)
- func TouchesNodeFields(updates map[string]any) bool
- type BalanceResult
- type Entitlement
- type EntitlementResult
- type GrantResult
- type GroupGrant
- type MembershipResult
- type Period
- type PlanSnapshot
- type ResetResult
- type TrafficEntry
- type TrafficResult
Constants ¶
const ( BalanceAccount = "balance" BalanceCommission = "commission_balance" )
Balance kinds.
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.
const ChangeRetention = 7 * 24 * time.Hour
ChangeRetention is how long the change log keeps rows; a consumer whose cursor is older lists again.
const RequestRetention = 90 * 24 * time.Hour
RequestRetention is how long the request ledger keeps a request id; a retry after that applies again.
Variables ¶
var ErrInsufficientBalance = errors.New("insufficient balance")
ErrInsufficientBalance means a debit would take a balance below zero.
var ErrMembershipNotFound = errors.New("subscription group membership not found")
ErrMembershipNotFound means the subscriber does not hold the group.
var ErrNegativeTraffic = errors.New("traffic must not be negative")
ErrNegativeTraffic means a traffic entry would decrease a counter.
var ErrSubscriberNotFound = errors.New("subscriber not found")
ErrSubscriberNotFound means no v2_user row has the id.
var ErrSubscriptionGroupNotFound = errors.New("subscription group not found")
ErrSubscriptionGroupNotFound means no v2_subscription_group row has the id.
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 ¶
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 ForgetActivityTx ¶ added in v4.2.0
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
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 ¶
PruneChanges deletes change log rows older than the retention.
func PruneRequests ¶
PruneRequests deletes request ledger rows older than RequestRetention.
func RecordChangesTx ¶
RecordChangesTx appends change log rows for the subscribers in tx, in the given order.
func RecordNodeGroupChangeTx ¶
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
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 ¶
Replay loads the recorded result of requestID into result and reports whether it was applied before.
func ResetActivityThrottleForTest ¶ added in v4.2.0
ResetActivityThrottleForTest forgets which users were written on db, so a test can record the same user again within the interval.
func TouchesNodeFields ¶
TouchesNodeFields reports whether a v2_user update changes a column that node user lists depend on.
Types ¶
type BalanceResult ¶
BalanceResult is the outcome of AdjustBalanceTx.
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 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 ¶
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 ¶
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.