mailbox

Documentation

Overview

Package mailbox is the central instance's message queue: a mailbox per person for the local agents that person federates, in the beekeeper database of the platform's Postgres.

Delivery is at least once and ordered per sender and receiver: a message stays until its receiver acks it or its deadline passes. A mailbox holds at most Cap unacked messages, and a sender's message id is accepted once within DedupeWindow. Every change of a mailbox is announced on the Postgres channel Channel, with the mailbox as its payload, so every instance learns of it.

Index

• Constants
• Variables
• func Key(person string) string
• type Delivery
• type Expired
• type Message
• type Sent
• type Store
  • func Open(ctx context.Context, url string) (*Store, error)
  • func (s *Store) Ack(ctx context.Context, mailbox string, seqs []int64, now time.Time) (int, error)
  • func (s *Store) Close()
  • func (s *Store) Expire(ctx context.Context, now time.Time) ([]Expired, error)
  • func (s *Store) Listen(ctx context.Context, changed func(mailbox string), ready func(), ...)
  • func (s *Store) Pending(ctx context.Context, mailbox string, now time.Time) (int, error)
  • func (s *Store) Receive(ctx context.Context, mailbox string, limit int, now time.Time) ([]Delivery, error)
  • func (s *Store) Send(ctx context.Context, m Message, now time.Time) (Sent, error)

Constants

const (
	// Cap is how many unacked messages a mailbox holds; the next is refused.
	Cap = 50
	// TTL is a message's life without a deadline of its own.
	TTL = 24 * time.Hour
	// DedupeWindow is how long a sender's message id is remembered.
	DedupeWindow = 24 * time.Hour
	// Channel is the Postgres channel a mailbox's changes are announced on.
	Channel = "beekeeper_mailbox"
)

const (
	KindMessage = "message"
	KindExpired = "expired"
)

The kinds of message: a sender's, and the notice to a sender that its message expired unacked.

Variables

var ErrFull = errors.New("the mailbox is full")

ErrFull is the refusal of a message to a mailbox that holds Cap.

Functions

func Key

func Key(person string) string

Key is a person's mailbox key: the email, lower-cased.

Types

type Delivery

type Delivery struct {
	// Seq is the delivery's number, the order of the mailbox: what an ack
	// names.
	Seq       int64           `json:"id"`
	Kind      string          `json:"kind"`
	Sender    string          `json:"sender"`
	MessageID string          `json:"messageId"`
	Envelope  json.RawMessage `json:"envelope"`
	Sent      time.Time       `json:"sent"`
	Deadline  time.Time       `json:"deadline"`
}

Delivery is a message in a mailbox.

type Expired

type Expired struct {
	Seq        int64     `json:"id"`
	Mailbox    string    `json:"mailbox"`
	Sender     string    `json:"sender"`
	SenderTeam string    `json:"senderTeam"`
	MessageID  string    `json:"messageId"`
	Deadline   time.Time `json:"deadline"`
}

Expired is a sender's message that expired unacked.

type Message

type Message struct {
	// Mailbox is the receiving person's email.
	Mailbox string
	// Sender is the sending person's email, SenderTeam the person's team.
	Sender     string
	SenderTeam string
	// ID is the sender's message id (the A2A messageId), unique per sender.
	ID string
	// Envelope is the message as the receiver gets it.
	Envelope json.RawMessage
	// Deadline is when it expires unacked; zero: TTL after it is sent.
	Deadline time.Time
}

Message is a message to send.

type Sent

type Sent struct {
	Seq       int64 `json:"id"`
	Duplicate bool  `json:"duplicate"`
}

Sent is the outcome of a send: the delivery's number, and whether the message id was seen before (nothing was queued again).

type Store

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

Store is the mailboxes in Postgres.

func Open

func Open(ctx context.Context, url string) (*Store, error)

Open connects to the database url names and migrates its schema.

func (*Store) Ack

func (s *Store) Ack(ctx context.Context, mailbox string, seqs []int64, now time.Time) (int, error)

Ack marks mailbox's deliveries seqs received and returns how many it acked; a number not pending in mailbox is none of its.

func (*Store) Close

func (s *Store) Close()

Close closes the connections.

func (*Store) Expire

func (s *Store) Expire(ctx context.Context, now time.Time) ([]Expired, error)

Expire takes the messages past their deadline out of their mailboxes, puts a notice of each sender's into the sender's mailbox (kind expired, outside the cap), forgets the acked messages and the message ids older than DedupeWindow, and returns the senders' expired messages.

func (*Store) Listen

func (s *Store) Listen(ctx context.Context, changed func(mailbox string), ready func(), lost func(error))

Listen calls changed with the mailbox of every change announced on Channel until ctx ends, reconnecting after a lost connection. ready, if set, is called each time the listener is in place.

func (*Store) Pending

func (s *Store) Pending(ctx context.Context, mailbox string, now time.Time) (int, error)

Pending is how many unacked, unexpired deliveries mailbox holds.

func (*Store) Receive

func (s *Store) Receive(ctx context.Context, mailbox string, limit int, now time.Time) ([]Delivery, error)

Receive returns the oldest unacked, unexpired deliveries of mailbox, at most limit, in order.

func (*Store) Send

func (s *Store) Send(ctx context.Context, m Message, now time.Time) (Sent, error)

Send queues m. A message id the sender used within DedupeWindow is not queued again: the first delivery's number comes back, marked a duplicate. A mailbox that holds Cap refuses with ErrFull.