paymentrequest

package
v1.38.0 Latest Latest
Warning

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

 Go to latest Published: Oct 30, 2026 License: MIT Imports: 32 Imported by: 0

Documentation

Overview

Package paymentrequest implements payment requests (see paymentrequestpb.PaymentRequestServer): a creator asking others for money.

Requests are layered, as the proto is. The first layer is where a request is made: today only a group chat (paymentrequestpb.GroupChatPaymentRequest), never a DM. The second is its type within that: today only a split (paymentrequestpb.Split), in which the creator asks named members of the chat (its debtors) each to pay them a share, all in one fiat currency. A kind or type added later is a new arm beside these (see PaymentRequest), not a rewrite of what is here.

A group chat request belongs to the chat it was created for, and is shown there as a message carrying a messagingpb.PaymentRequestWidget, which names the request by ID alone: the shares and state come from this package's RPCs. It is live from creation, sent or not, and may be sent any number of times. A split's title and description are plaintext, and moderated, in a group that is not private, and encrypted with the chat key in a private group (see paymentrequestpb.EncryptedPaymentRequestText); everything else (the currency, ends_at, the shares and how each was paid) is plaintext on the server in either, since the server checks every payment against its share and decides when the request is complete.

A request is OPEN until one of three things ends it, each final:

  • every share is paid (COMPLETED). A share is paid all at once, for exactly its amount, either by a Flipcash payment (paymentrequestpb.PaidShare.Intent), at most once (see PaymentRequest.PaymentIntentIDs), or by the creator marking it paid by hand (PaidShare.Manual); the creator's own share is paid from creation (PaidShare.Creator). A paid share stays paid, and only a hand mark is ever replaced: by a payment that was in flight when it was made (see PaidShareSupersedes).
  • the creator closes it (CLOSED).
  • its ends_at passes (EXPIRED). ends_at is the request's one hard deadline, at most MaxDuration after its creation: no share can be paid after it. Expiry is decided by the clock (see PaymentRequest.StatusAt), and recorded later by a worker only so that listings can file the request as resolved (see ExpireDelay).

Nothing about a request is editable after creation: not its text, currency, deadline nor shares. No event announces a change to a request; clients refetch it, ordering copies by PaymentRequest.Version.

Index

Constants

View Source 
const ExpireDelay = time.Minute

ExpireDelay is how long after a request's ends_at the expiry worker takes it up (see Store.GetDueForExpiry and Store.Expire), as poll.TallyDelay is for a poll's tally.

Nothing waits on the worker: from ends_at on, a request stored OPEN is EXPIRED to every reader by the clock (see PaymentRequest.StatusAt), and no share can be marked paid. The worker only records it, so listings can file it as resolved. The delay is not needed for correctness either: a payment validated just before ends_at is recorded by a task moments later (see Store.PayShare), and that write and the worker's are each conditioned on the version they read, so they order themselves whichever lands first: a payment recorded first may complete the request, and one recorded after the expiry is still recorded on the EXPIRED request. The delay only keeps that race from being the common case, so a request whose last share is paid at its deadline usually reads COMPLETED rather than EXPIRED.

View Source 
const IDSize = 16

IDSize is the length, in bytes, of a PaymentRequestId.

View Source 
const IdempotencyKeySize = 16

IdempotencyKeySize is the length, in bytes, of a client's IdempotencyKey.

View Source 
const MaxDuration = 4 * 7 * 24 * time.Hour

MaxDuration bounds how long after its creation a request may end (see Split.ends_at): four weeks, the longest a request is expected to stay open. It also keeps every ends_at, and so every time the stores derive from it (the expiry queue's due time and the open section's sort key, both Unix nanos), well inside the range an int64 of nanoseconds can represent.

View Source 
const MaxExpiryQueueStatsDepth = 1000

MaxExpiryQueueStatsDepth bounds the walk behind Store.GetExpiryQueueStats: a backlog deeper than this reports this depth, so the gauges cost one bounded read even mid-incident.

View Source 
const PaymentIntentIDSize = 32

PaymentIntentIDSize is the length, in bytes, of a share's payment intent ID (see PaymentRequest.PaymentIntentIDs): an OCP intent ID.

Variables

View Source 
var (
	// ErrNotFound is returned for a payment request that does not exist.
	ErrNotFound = errors.New("payment request not found")

	// ErrExists is returned by Create when a request with the same ID already
	// exists: an earlier attempt of the same CreatePaymentRequest created it
	// (see MustDeriveID).
	ErrExists = errors.New("payment request already exists")

	// ErrNotAShare is returned by PayShare when no share of the request names
	// the user.
	ErrNotAShare = errors.New("no share for user")

	// ErrNotOpen is returned by PayShare for a share marked paid by hand while
	// the request is not open as of now: stored completed, closed or expired,
	// or stored open but past its ends_at (see PaymentRequest.IsOpenAt).
	ErrNotOpen = errors.New("payment request is not open")

	// ErrInvalidPaidShare is returned by PayShare for a PaidShare it cannot
	// record: one without paid_at, or one paid other than by a payment or by
	// hand. A creator's share is paid from creation (see NewSplitRequest), so
	// PaidShare.Creator never reaches PayShare.
	ErrInvalidPaidShare = errors.New("invalid paid share")

	// ErrWrongIntent is returned by PayShare for a payment naming an intent
	// other than the one the share's payment must use (see
	// PaymentRequest.PaymentIntentIDs), the creator's own share included,
	// which no payment pays.
	ErrWrongIntent = errors.New("payment intent is not the share's")
)
View Source 
var ErrInvalidShares = errors.New("invalid shares")

ErrInvalidShares is wrapped by every refusal of a split's shares (see NewSplitRequest and PaymentRequest.Validate), so a caller can answer them all as INVALID_SHARES.

Functions

func ExpiryDueAt 

func ExpiryDueAt(r *PaymentRequest) time.Time

ExpiryDueAt is when a request first comes due for the expiry worker: ExpireDelay after its ends_at.

func MustDeriveID 

func MustDeriveID(creatorID *commonpb.UserId, key *paymentrequestpb.IdempotencyKey) *paymentrequestpb.PaymentRequestId

MustDeriveID returns the ID of the payment request that a CreatePaymentRequest from creatorID carrying key creates.

Like a poll's ID (see poll.MustDerivePollID), a request's ID is derived from its creator and the request's idempotency key rather than minted at random, so the ID is itself the idempotency record: a retry derives the same ID, and the store's uniqueness condition turns the duplicate into a read of the original, whatever kind or parameters the retry carries. The creator is part of the input so no client can derive another user's request ID. The digest is truncated to IDSize bytes and stamped as a version 8 UUID, as a group ID is.

It panics if either input is not its fixed width, which would be a programming error: user IDs are UUIDs, and a request's key is checked to width before it reaches here.

func PaidShareSupersedes 

func PaidShareSupersedes(current, next *paymentrequestpb.PaidShare) bool

PaidShareSupersedes reports whether Store.PayShare records next over a share's current paid (nil while unpaid): always for an unpaid share, and for a share marked paid by hand when next is a payment, never otherwise.

A hand mark is the creator's word that a share was settled some other way; a payment is the money itself. A creator can mark a share by hand while its debtor's payment is in flight (validated against the unpaid share, not yet recorded by its task), so a payment landing after a hand mark replaces it, and the record shows what was paid. Nothing replaces a payment, nor the creator's own share, and a hand mark never replaces anything.

func PublishClosed added in v1.38.0 

func PublishClosed(chatEventBus *event.Bus[*commonpb.ChatId, *eventpb.ChatEvent], r *PaymentRequest, now time.Time)

PublishClosed announces the close of r, as Store.Close returned it from a write that changed it.

func PublishCreated added in v1.38.0 

func PublishCreated(chatEventBus *event.Bus[*commonpb.ChatId, *eventpb.ChatEvent], r *PaymentRequest, now time.Time)

PublishCreated announces r, just created, whole, as a reader sees it at now.

func PublishSharePaid added in v1.38.0 

func PublishSharePaid(chatEventBus *event.Bus[*commonpb.ChatId, *eventpb.ChatEvent], r *PaymentRequest, userID *commonpb.UserId, completed bool, now time.Time)

PublishSharePaid announces the payment, or marking, of userID's share of r, as Store.PayShare returned it from a write that changed it: the share as it now stands and, when the write completed the request (PayShare's completed), the request's completion.

func ValidatePaidShare 

func ValidatePaidShare(paid *paymentrequestpb.PaidShare) error

ValidatePaidShare checks that paid is one Store.PayShare records: it has a paid_at, and it is paid by a payment naming its intent or marked paid by hand. It returns an error wrapping ErrInvalidPaidShare otherwise. Every store checks it before reading anything.

func ValidatePaymentIntent 

func ValidatePaymentIntent(r *PaymentRequest, userID *commonpb.UserId, paid *paymentrequestpb.PaidShare) error

ValidatePaymentIntent checks that a payment of userID's share of r names the intent that share's payment must use (see PaymentRequest.PaymentIntentIDs), returning ErrWrongIntent otherwise. A hand mark names no intent and passes. Every store checks it once the share is found, before deciding anything else, so a share already paid refuses a wrong intent too.

It is not what keeps a share from being paid twice: that is intent validation, which refuses a payment with any other intent ID before money moves, and OCP, which stores an intent ID once (see PaymentIntentIDs). By the time a payment reaches the store its money has moved, so this check cannot stop it. It keeps the record honest, so a share is only ever recorded as paid by its own intent, and it makes any path that reaches the store without that validation fail loudly instead of passing as a no-op on a share already paid.

Types

type ExpiryQueueStats 

type ExpiryQueueStats struct {
	// DueDepth is how many queued requests are due, at most
	// MaxExpiryQueueStatsDepth.
	DueDepth uint64

	// OldestDueAt is the due time of the most overdue request, or the zero
	// time when none is due.
	OldestDueAt time.Time
}

ExpiryQueueStats is a snapshot of the expiry worker's backlog, backing its gauges (see Store.GetExpiryQueueStats). Only due requests count: one waiting for its ends_at is not work the worker is behind on. A claimed request is not due until its lease ends, and one whose attempt failed not until its retry, so a request stuck retrying shows here only between its retry coming due and a worker claiming it; its failures are logged by the worker.

type ExpiryTask 

type ExpiryTask struct {
	ID *paymentrequestpb.PaymentRequestId

	// Attempts is the number of failed expiry attempts recorded so far (via
	// Store.DelayExpiry). The worker uses it to pace backoff.
	Attempts uint32

	// NextAttemptAt is when the task next becomes due: ExpireDelay after the
	// request's ends_at at first (see ExpiryDueAt), then whenever a claim or a
	// delay moves it.
	NextAttemptAt time.Time
}

ExpiryTask is a queued unit of expiry work: a request stored open, with the retry bookkeeping the worker schedules off of (see Store.GetDueForExpiry).

type PageCursor 

type PageCursor struct {
	At time.Time
	ID []byte
}

PageCursor is a position in one section of a chat's requests: the sort time (ends_at in the open section, ClosedAt in the resolved one) and ID of the last request read. The ID makes the order total, so a page resumes strictly after the request the cursor names even when others share its time.

func CursorOf 

func CursorOf(r *PaymentRequest, section Section) *PageCursor

CursorOf returns the cursor naming r's position in section.

type PaymentRequest 

type PaymentRequest struct {
	ID        *paymentrequestpb.PaymentRequestId
	Creator   *commonpb.UserId
	CreatedAt time.Time

	// ChatID is the group chat the request was made in.
	ChatID *commonpb.ChatId

	// Split is set for a split, the only type today.
	Split *paymentrequestpb.Split

	// PaymentIntentIDs holds, for every debtor's share (every share but the
	// creator's own), the OCP intent ID a payment of it must use, keyed by
	// the debtor's user ID (the string of its bytes). Each is
	// PaymentIntentIDSize random bytes, drawn once at creation (see
	// NewSplitRequest) and never changed.
	//
	// It is what makes a share payable at most once. OCP stores an intent
	// ID at most once, in the same transaction that moves the money, so a
	// payment refused for any ID but this one (see intent.Integration) can
	// commit once for the share, however many times or from however many
	// devices the debtor tries. The ID is the server's, not derived from
	// anything a client sees: an ID anyone could compute could be used up by
	// someone else first, leaving the share unpayable.
	//
	// It is a secret until used, so it is never projected (see ToProto): the
	// GetPaymentRequest RPC gives a debtor their own while their share is
	// unpaid, and no one else. Once the share is paid by it, the ID is
	// public in PaidShare.intent, which costs nothing, since it can never be
	// used again.
	PaymentIntentIDs map[string][]byte

	Status   Status
	Version  uint64
	ClosedAt *time.Time
}

PaymentRequest is a payment request as stored.

What every kind shares is held directly: ID, Creator, CreatedAt and the state (Status, Version, ClosedAt). The layers are held one field per arm, as the proto's oneofs are: ChatID is the first layer, a group chat request (paymentrequestpb.GroupChatPaymentRequest; a 16-byte group chat ID, never a DM's), and Split is the type within it. Exactly one type is set. Another kind or type is a new field beside these, with a case in EndsAt, Validate and ToProto, and nothing else changes.

Split is exactly what the creator chose, except each share's paid, the only part of a request that changes after creation (see Store.PayShare): a share is addressed by its user ID (see Share), and once paid stays paid, changing only from a hand mark to a payment (see PaidShareSupersedes). The creator's own share, when they listed one, is paid from creation (paymentrequestpb.PaidShare.Creator with paid_at = CreatedAt; see NewSplitRequest). Its text is plaintext in a group that is not private and encrypted in a private group; the store holds either as given.

Version starts at 1 on creation and advances by exactly one on every change (a share paid, by payment or by hand, a completion, a close); an idempotent no-op leaves it alone, and so does expiry, which is the clock's (see paymentrequestpb.PaymentRequestMetadata). ClosedAt is nil while Status is StatusOpen, and set once it is not: ends_at when expired, when the creator closed it when closed, when its last share was paid when completed.

func NewSplitRequest 

func NewSplitRequest(
	id *paymentrequestpb.PaymentRequestId,
	creator *commonpb.UserId,
	chatID *commonpb.ChatId,
	split *paymentrequestpb.Split,
	createdAt time.Time,
) (*PaymentRequest, error)

NewSplitRequest returns the split request that creator, in the group chat chatID, asks for with split at createdAt: stored open at version 1, with the creator's own share, if they listed one, paid from creation (paymentrequestpb.PaidShare.Creator at createdAt), and a random payment intent ID for every other share (see PaymentRequest.PaymentIntentIDs).

split is as a client sends it, which it does not modify: an error wrapping ErrInvalidShares is returned when a share has paid set, two shares name the same user, or no share names anyone but the creator (see Validate for the rest). What it cannot judge alone is the caller's: that ends_at is in the future and at most MaxDuration away, that every debtor is a member of the chat, that each amount is a whole number, at least one, of the currency's minimum transfer value, and that the currency can be valued.

func (*PaymentRequest) AllSharesPaid 

func (r *PaymentRequest) AllSharesPaid() bool

AllSharesPaid reports whether every share is paid, which is what completes an open request.

func (*PaymentRequest) Clone 

func (r *PaymentRequest) Clone() *PaymentRequest

Clone returns a deep copy of the request.

func (*PaymentRequest) ClosedAtAt 

func (r *PaymentRequest) ClosedAtAt(now time.Time) (closedAt time.Time, ok bool)

ClosedAtAt is when the request stopped being open, as a reader is shown it at now: the stored ClosedAt, or ends_at for a request expired by the clock alone. ok is false while the request is open at now.

func (*PaymentRequest) CreatorShare 

func (r *PaymentRequest) CreatorShare() *paymentrequestpb.Share

CreatorShare returns the creator's own share, paid from creation, or nil when they listed none.

func (*PaymentRequest) EndsAt 

func (r *PaymentRequest) EndsAt() time.Time

EndsAt is when the request stops accepting payments: its type's deadline (for a split, Split.ends_at).

func (*PaymentRequest) IsOpenAt 

func (r *PaymentRequest) IsOpenAt(now time.Time) bool

IsOpenAt reports whether the request accepts a share being marked paid, or closed, at now: it is stored open and has not reached its ends_at.

func (*PaymentRequest) PaymentIntentID 

func (r *PaymentRequest) PaymentIntentID(userID *commonpb.UserId) []byte

PaymentIntentID returns the intent ID a payment of userID's share must use (see PaymentIntentIDs), or nil when they hold no share a payment can pay: no share at all, or the creator's own. It is the stored value: callers must not modify it.

func (*PaymentRequest) Share 

func (r *PaymentRequest) Share(userID *commonpb.UserId) *paymentrequestpb.Share

Share returns the share naming userID, or nil when none does. It is the stored value, which only a store modifies (see Store.PayShare).

func (*PaymentRequest) Shares 

func (r *PaymentRequest) Shares() []*paymentrequestpb.Share

Shares returns the request's shares, in the order the creator listed them. They are the stored values: callers must not modify them.

func (*PaymentRequest) StatusAt 

func (r *PaymentRequest) StatusAt(now time.Time) Status

StatusAt is the status a reader is shown at now: the stored status, except that a request stored open is expired from its ends_at on, since nothing can be paid from then (see ExpireDelay).

func (*PaymentRequest) ToProto 

func (r *PaymentRequest) ToProto(now time.Time) *paymentrequestpb.PaymentRequestMetadata

ToProto projects the request onto paymentrequestpb.PaymentRequestMetadata as a reader sees it at now: a request stored open past its ends_at is EXPIRED, closed at its ends_at, at the version it was stored with (see StatusAt and ClosedAtAt), which is exactly what the expiry worker will store, so what a reader is shown never changes when it does. It never carries PaymentIntentIDs, which only their debtor is given (see PaymentIntentIDs).

func (*PaymentRequest) Validate 

func (r *PaymentRequest) Validate() error

Validate checks that the request is one a store may hold: every ID its fixed width, exactly one type set, and for a split an ends_at and shares that name distinct users, at least one of them not the creator, with the creator's own share (if any) paid from creation and no other share claiming to be, and a payment intent ID for every share but the creator's and for nothing else. Share problems wrap ErrInvalidShares.

It checks shape only: nothing about the chat, the currency, the amounts or the clock, which are the creating caller's (see NewSplitRequest).

type Section 

type Section uint8

Section is one of the two sections a chat's requests are paged in (see Store.GetChatPage and paymentrequestpb.GetChatPaymentRequestsRequest_Section). 

const (
	// SectionOpen holds the requests stored open, ordered by ends_at
	// ascending, then ID ascending.
	SectionOpen Section = iota + 1

	// SectionResolved holds the requests stored completed, closed or expired,
	// ordered by ClosedAt descending, then ID descending.
	SectionResolved
)

func (Section) String 

func (s Section) String() string

type Server 

type Server struct {
	paymentrequestpb.UnimplementedPaymentRequestServer
	// contains filtered or unexported fields
}

Server implements paymentrequestpb.PaymentRequestServer.

Who may do what in a group chat request is the chat's to say, through the shared chat.Access, except where the request itself names who:

  • Creating a request is speaking in the chat (Access.SpeakerStanding), and the standing also says whether the chat takes a split's text in plaintext (a public group) or encrypted under the chat key (a keyed private group). For now public groups are refused outright (see CreatePaymentRequest), a transitional gate.
  • Seeing a request, alone or in its chat's listing, is reading the chat in full (Access.CanListen): its members, and a non-member of a public group who satisfies its listener rules.
  • Closing a request and marking a share paid by hand are its creator's alone, whatever their standing in the chat now: the request is theirs, and they are the one who is paid.

Paying a share with a Flipcash payment is not this service's: it is an intent, validated and recorded elsewhere (see Store.PayShare).

Every change the service makes, a creation, a close or a share marked paid by hand, is announced on the chat's topic (see PublishCreated); a no-op is not.

func NewServer 

func NewServer(
	log *zap.Logger,

	authz auth.Authorizer,

	requests Store,
	chats chat.Store,
	access *chat.Access,
	moderator moderation.Client,

	chatEventBus *event.Bus[*commonpb.ChatId, *eventpb.ChatEvent],

	opts ...ServerOption,
) *Server

NewServer returns a PaymentRequest service over requests. access is the chat.Access the chat and messaging servers share, so a request is gated by exactly what gates the chat it belongs to; chats is read for what Access does not answer (a group's existence, and whether a split's debtors are its members). chatEventBus is the chat-keyed event bus every change is announced on (see PublishCreated).

func (*Server) ClosePaymentRequest 

func (s *Server) ClosePaymentRequest(ctx context.Context, req *paymentrequestpb.ClosePaymentRequestRequest) (*paymentrequestpb.ClosePaymentRequestResponse, error)

ClosePaymentRequest closes an open request. A request that does not exist is NOT_FOUND, and only its creator may close it (DENIED), whatever their standing in its chat now. Closing a request that is not open, stored final or past its ends_at, is the no-op OK (see Store.Close), answered with the request as it is. A real close is announced on the chat's topic (see PublishClosed).

func (*Server) CreatePaymentRequest 

func (s *Server) CreatePaymentRequest(ctx context.Context, req *paymentrequestpb.CreatePaymentRequestRequest) (*paymentrequestpb.CreatePaymentRequestResponse, error)

CreatePaymentRequest creates a payment request. Today the only kind is a split in a group chat.

The request is checked in the order a client can act on, local checks first, the store and chat reads next, and moderation, a call out, last: the chat must be a group (CHAT_NOT_SUPPORTED); the caller must be able to speak in it (DENIED); it must not be a public group (CHAT_NOT_SUPPORTED, transitional, see below); the text must be in the form the chat takes, plaintext in a public group and encrypted under the chat key in a private one (ENCRYPTION_NOT_ALLOWED, ENCRYPTION_REQUIRED); ends_at must be in the future and at most MaxDuration away (INVALID_ENDS_AT); the currency must be one the server values (INVALID_CURRENCY); the shares must be valid (INVALID_SHARES, see validateShares); and plaintext must pass moderation (TEXT_MODERATED, see moderate). Only then is anything written.

A chat that does not exist is DENIED, not reported: the proto has no NOT_FOUND for creation, and Access.SpeakerStanding admits no one to a chat nothing is stored under, so a missing group is refused exactly as a group the caller is not in.

The proto supports public groups; they are refused for now, a transitional gate removed by deleting this check (see WithPublicGroupRequests). The plaintext path below is built and tested so lifting it changes nothing else.

The RPC is retry-safe. The request's ID is derived from the caller and the request's idempotency key (see MustDeriveID), so a retry names the same request, and one that already exists is answered from the store before any check runs, whatever the retry carries: a member since departed or a text the moderator has since learned to flag are facts about a new request, not this one. A retry that loses a race with its twin is caught by the store's uniqueness condition and answered the same way.

Only the write that creates the request announces it (see PublishCreated); a retry answered from the store announces nothing, its twin having done so.

func (*Server) GetChatPaymentRequests 

func (s *Server) GetChatPaymentRequests(ctx context.Context, req *paymentrequestpb.GetChatPaymentRequestsRequest) (*paymentrequestpb.GetChatPaymentRequestsResponse, error)

GetChatPaymentRequests pages one section of a group chat's requests (see Store.GetChatPage for each section's order).

A DM is DENIED before anything is read, a token that is malformed or bound to another chat or section is InvalidArgument, a group that does not exist is NOT_FOUND, and the caller must be able to read it in full (DENIED). The group's existence is told apart by reading its rules (Store.GetGroupRules, which a store caches for the private groups requests live in) and then asking Access with them, since Access alone admits no one to a missing chat without saying so.

The open section is by stored status, so it can hold a request past its ends_at that the expiry worker has not yet stored expired. Such a request is EXPIRED to every reader and is dropped from the page, and the store is read on until the page is full or it runs dry, so a backlog of them never yields an empty page that claims more.

func (*Server) GetPaymentRequest 

func (s *Server) GetPaymentRequest(ctx context.Context, req *paymentrequestpb.GetPaymentRequestRequest) (*paymentrequestpb.GetPaymentRequestResponse, error)

GetPaymentRequest returns a request as a reader sees it now.

A request that does not exist is NOT_FOUND, checked before the caller's standing; a request ID is derived from its creator and a key only they hold, so the distinction leaks nothing. Otherwise the caller must be able to read the request's chat in full (DENIED).

A caller holding an unpaid share of the request is also given the intent ID their payment of it must use (PaymentRequest.PaymentIntentIDs), whatever the request's status: a payment's validation judges that. No one else is ever given it, the creator included.

func (*Server) MarkPaid 

func (s *Server) MarkPaid(ctx context.Context, req *paymentrequestpb.MarkPaidRequest) (*paymentrequestpb.MarkPaidResponse, error)

MarkPaid marks a debtor's share paid by hand (PaidShare.Manual). A request that does not exist is NOT_FOUND, only its creator may mark a share (DENIED), and the request must have a share for the user (NOT_A_SHARE). A share already paid, by payment, by hand or from creation, is the no-op OK whatever the request's status; an unpaid share of a request that is not open is NOT_OPEN (see Store.PayShare). No message is posted; a real change is announced on the chat's topic (see PublishSharePaid).

type ServerOption 

type ServerOption func(*Server)

ServerOption configures a Server at construction beyond its required dependencies.

func WithClock 

func WithClock(now func() time.Time) ServerOption

WithClock overrides the clock the server decides a request's status and deadlines by, for tests.

func WithPublicGroupRequests 

func WithPublicGroupRequests() ServerOption

WithPublicGroupRequests lifts the transitional refusal of requests in public groups (see CreatePaymentRequest), so the plaintext path stays built and tested while it is closed. It is for tests only: no production server sets it, and opening public groups is deleting the check, not setting this.

type Status 

type Status uint8

Status is a request's status. It only moves forward: from open to one of completed, closed or expired, which are final.

All four are stored, but expired is also the clock's: a request stored open is expired to every reader from its ends_at (see PaymentRequest.StatusAt), and the expiry worker stores it later (see ExpireDelay and Store.Expire). 

const (
	StatusOpen Status = iota
	StatusCompleted
	StatusClosed
	StatusExpired
)

func (Status) IsFinal 

func (s Status) IsFinal() bool

IsFinal reports whether s is one a request never leaves: anything but open.

func (Status) String 

func (s Status) String() string

func (Status) ToProto 

func (s Status) ToProto() paymentrequestpb.PaymentRequestStatus

ToProto projects the status onto paymentrequestpb.PaymentRequestStatus.

type Store 

type Store interface {
	// Create writes a new request as given, with status open, version 1 and no
	// ClosedAt whatever r says, and enqueues it for expiry, due at
	// ExpiryDueAt. A request failing PaymentRequest.Validate is refused with
	// its error. ErrExists is returned, and nothing is written, when a request
	// with its ID already exists.
	Create(ctx context.Context, r *PaymentRequest) error

	// Get returns a request, or ErrNotFound.
	Get(ctx context.Context, id *paymentrequestpb.PaymentRequestId) (*PaymentRequest, error)

	// GetChatPage returns up to limit of the requests made in the group chat
	// chatID in one section, in that section's order, strictly after cursor
	// (from the start when nil): the open section (stored open) by ends_at
	// ascending then ID ascending, and the resolved section (stored completed,
	// closed or expired) by ClosedAt descending then ID descending (see
	// CursorOf). Whether more remain is the caller's to judge, by asking for
	// one more than it shows.
	//
	// Sections are by stored status, so a request stored open past its ends_at
	// is still in the open section until the expiry worker stores it expired,
	// about ExpireDelay later, and only then in the resolved one. The reader
	// drops it from an open page by PaymentRequest.StatusAt and so, for that
	// moment, shows it in neither; the lag is accepted, since a client already
	// judges an open request past its ends_at as expired itself.
	GetChatPage(ctx context.Context, chatID *commonpb.ChatId, section Section, cursor *PageCursor, limit int) ([]*PaymentRequest, error)

	// PayShare records userID's share of a request as paid by paid, as of now,
	// and returns the request as it then is, reporting whether anything
	// changed. ErrNotFound is returned for a request that does not exist,
	// ErrNotAShare when no share names userID, ErrInvalidPaidShare for a paid
	// it cannot record, and ErrWrongIntent for a payment whose intent is not
	// the share's (see ValidatePaymentIntent), whether or not the share is
	// already paid.
	//
	// A share already paid is a no-op, whatever the request's status, except
	// that a payment replaces a hand mark (see PaidShareSupersedes): a
	// repeated payment task, or a creator marking by hand a share already
	// paid, changes nothing. A share has one intent that can pay it, so the
	// only payment it records is that intent's, once. Otherwise the rule
	// depends on how it is paid:
	//   - by hand (PaidShare.Manual): only while the request is open as of now
	//     (see PaymentRequest.IsOpenAt); ErrNotOpen otherwise.
	//   - by a payment (PaidShare.Intent): whatever the status. The money has
	//     moved, and the payment was validated while the request was open, so
	//     a close or an expiry, or a hand mark, that lands between the
	//     validation and this write must not lose it.
	//
	// A real change sets the share's paid and advances the version by one. If
	// it paid the last unpaid share of a request stored open, the request is
	// completed in the same write, with ClosedAt paid.paid_at, and leaves the
	// expiry queue, and completed is reported; a request already final keeps
	// its status and ClosedAt. A payment replacing a hand mark pays no share
	// that was unpaid, so it never completes a request, and leaves the status
	// and ClosedAt as they were, even when the hand mark completed it. The
	// request returned cannot tell those two apart (both are COMPLETED, the
	// share paid by a payment), which is why completed is reported: it is what
	// announcing the change needs (see PublishSharePaid).
	PayShare(ctx context.Context, id *paymentrequestpb.PaymentRequestId, userID *commonpb.UserId, paid *paymentrequestpb.PaidShare, now time.Time) (r *PaymentRequest, changed, completed bool, err error)

	// Close closes a request open as of now: status closed, ClosedAt now,
	// version advanced by one, and off the expiry queue. A request that is not
	// open as of now, because it is stored final or has reached its ends_at,
	// is a no-op. Either way it returns the request as it then is, and reports
	// whether it changed. ErrNotFound is returned for a request that does not
	// exist. Who may close is the caller's.
	Close(ctx context.Context, id *paymentrequestpb.PaymentRequestId, now time.Time) (r *PaymentRequest, changed bool, err error)

	// GetDueForExpiry returns up to limit requests on the expiry queue whose
	// due time is at or before asOf, soonest first. The queue is eventually
	// consistent: a request just claimed or made final may still be listed,
	// which ClaimForExpiry settles.
	GetDueForExpiry(ctx context.Context, asOf time.Time, limit int) ([]*ExpiryTask, error)

	// GetExpiryQueueStats reports the expiry backlog as of asOf: how many
	// queued requests are due, counting at most MaxExpiryQueueStatsDepth, and
	// the due time of the most overdue. Requests not yet due are never read,
	// so its cost follows the backlog, not the number of open requests. It is
	// eventually consistent, like GetDueForExpiry.
	GetExpiryQueueStats(ctx context.Context, asOf time.Time) (*ExpiryQueueStats, error)

	// ClaimForExpiry pushes a queued request's due time out to until, provided
	// it is still queued and due as of asOf. It reports whether the claim was
	// performed: false means the request left the queue (it was made final)
	// or is no longer due (another server claimed or delayed it first).
	//
	// A claim is an efficiency guard, not a lock: Expire is idempotent and
	// safe to run concurrently, and claiming only keeps servers from
	// duplicating it. A crashed claimant needs no recovery: the claim expires
	// when its until passes and the request simply becomes due again.
	ClaimForExpiry(ctx context.Context, id *paymentrequestpb.PaymentRequestId, asOf, until time.Time) (bool, error)

	// DelayExpiry reschedules a queued request after a failed attempt: its due
	// time moves to nextAttemptAt and its failed-attempt count increments. It
	// is a no-op on a request that is no longer queued (it was made final,
	// dequeuing it).
	DelayExpiry(ctx context.Context, id *paymentrequestpb.PaymentRequestId, nextAttemptAt time.Time) error

	// Expire stores a request stored open as expired: status expired, ClosedAt
	// its ends_at, and off the expiry queue, with the version unchanged (see
	// Store), and reports whether it changed anything. A request already final
	// is a no-op (dequeued, should it still be queued). ErrNotFound is
	// returned for a request that does not exist. Whether it has reached its
	// ends_at is the caller's (see ExpireDelay).
	//
	// A payment recorded after the expiry is still recorded (see PayShare),
	// and the request stays expired.
	Expire(ctx context.Context, id *paymentrequestpb.PaymentRequestId) (changed bool, err error)
}

Store persists payment requests.

A request is one item, written once with everything its creator chose (see PaymentRequest), of which only each share's paid and the state change afterward. Every write of an existing request is a compare-and-set on its version inside the store, retried from a fresh read when it loses, so concurrent payments of different shares, a payment racing a close, and the expiry worker racing either all serialise on the one item and none is lost.

Status moves forward only: from open to completed, closed or expired, never back and never between the three. Every change advances the version by exactly one, and an idempotent no-op (a share already paid, a close of a request that is not open, an expiry of a final one) writes nothing and leaves it alone. The one write that changes a request without advancing its version is Expire: expiry is the clock's (see PaymentRequest.StatusAt), so storing it changes nothing a reader is shown.

Every request stored open is on an expiry queue (GetDueForExpiry, ClaimForExpiry, DelayExpiry), due ExpireDelay after its ends_at, and leaves it in the write that makes it final, whichever that is: the last share paid, a close, or Expire.

Reads of a request are strongly consistent: a creator reads their request right after creating it, and a payer theirs right after paying. A chat's listing (GetChatPage) and the expiry queue are eventually consistent, and need not be more: a listing is current only as of its read anyway (see the GetChatPaymentRequests RPC), and a claim re-checks the request it names.

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

Worker drains the expiry queue: it polls the store for requests stored open whose expiry is due (ExpireDelay after their ends_at) and stores each one expired (see Store.Expire).

It decides nothing a reader sees. Expiry is already the clock's: from its ends_at on, a request stored open is EXPIRED to every reader, closed at its ends_at, at the version it was stored with (see PaymentRequest.StatusAt and ToProto), and no share can be marked paid on it. Storing the expiry changes none of that; the worker exists only to move the request from its chat's open section to the resolved one, with closed_at = ends_at, so that the RESOLVED listing is one read of one index rather than a merge of the resolved requests and the open ones past their end (see Store.GetChatPage). A slow or stopped worker therefore delays only the listing: an expired request sits at the head of the open section, which a reader drops by StatusAt, until the worker files it, and never shows a state it would not show afterward.

Like poll's tally, a failed attempt is retried under backoff, forever: a request stored open past its end must eventually be filed as resolved, and a later attempt is no less correct than the first. A payment recorded between the worker's read and its write is never lost: both writes are conditioned on the version and status they read, so whichever lands second decides again (see ExpireDelay).

It implements the OCP worker.Runtime interface, so the parent application registers it alongside its other background runtimes and controls the poll interval. It is safe to run on every server instance: workers claim a request before expiring it, and Expire is conditional and idempotent, so overlap costs a duplicate read at worst.

Construct with NewWorker and either run it via Start (the OCP runtime entry point) or drive single ticks with Process (tests).

func NewWorker 

func NewWorker(log *zap.Logger, requests Store, opts ...WorkerOption) *Worker

NewWorker returns a Worker draining the expiry queue of requests.

func (*Worker) Process 

func (w *Worker) Process(runtimeCtx context.Context) (int, error)

Process runs one worker tick: it pulls the due requests and expires them with bounded concurrency, reporting how many this worker took on (claimed). Zero means the due queue is drained: requests other workers hold claims on are not counted.

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context, interval time.Duration) error

Start satisfies the OCP worker.Runtime interface: it polls the expiry queue every interval until ctx is cancelled, whose error it returns. The cadence is fixed-rate, not fixed-delay: each sleep is the interval minus the time the tick's processing took. A full batch polls again immediately, so a burst of requests ending together drains at processing speed rather than one batch per interval. It also runs the queue gauges (depth, max age) alongside the processing loop.

type WorkerOption 

type WorkerOption func(*Worker)

WorkerOption overrides one of the worker's tuning knobs.

func WithWorkerBackoff 

func WithWorkerBackoff(base, maxDelay time.Duration) WorkerOption

WithWorkerBackoff overrides the retry backoff's base and maximum delay.

func WithWorkerBatchSize 

func WithWorkerBatchSize(n int) WorkerOption

WithWorkerBatchSize overrides how many due requests one tick pulls from the queue.

func WithWorkerClaimLease 

func WithWorkerClaimLease(lease time.Duration) WorkerOption

WithWorkerClaimLease overrides how far a claim pushes a request's due time out.

func WithWorkerExpireTimeout 

func WithWorkerExpireTimeout(timeout time.Duration) WorkerOption

WithWorkerExpireTimeout overrides the bound on a single expiry attempt.

func WithWorkerMaxConcurrency 

func WithWorkerMaxConcurrency(n int) WorkerOption

WithWorkerMaxConcurrency overrides how many requests are expired at once.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.