Help us shape the future of Go! Take the 2026 Go Developer Survey

store

package
v0.5.0 Latest Latest
Warning

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

Go to latest
Published: Sep 28, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

Documentation

Overview

Package store persists an awp peer's state in SQLite: the outbox of unacked messages, the per-thread "last seen" ids that resume needs, a log of every message sent and received, thread state, blobs and grants.

Section 4.3 requires this to be on disk, because the process can be killed together with its sandbox at any moment. Every function that changes state is safe to interrupt: either the transaction commits or it never happened.

Index

Constants

View Source
const (
	RoleIssued    = "issued"    // we signed it
	RoleHeld      = "held"      // someone granted it to us
	RolePresented = "presented" // a peer showed it to us about itself
)

Grant roles.

Variables

View Source
var InboxSys = []string{"blob", "bye", "refused", "shares", "private"}

InboxSys lists the local (sys) event types that belong in an inbox.

Functions

func AckOutbox

func AckOutbox(q Q, peer, id string, now time.Time) (th, t string, found bool, err error)

AckOutbox removes the acked message from the outbox, together with any earlier never-acked lines in the same thread, and marks the outgoing blobs those lines carried as sent. It reports the row's thread and type, and whether it was found.

func AddAddr

func AddAddr(q Q, peer, addr, source string, now time.Time) error

AddAddr records an address hint. A hint from a more authoritative source upgrades the stored source.

func Append

func Append(q Q, r *Record) (inserted bool, err error)

Append adds a record, unless one with the same (peer, dir, id) exists, in which case it reports inserted == false. That is the receive-side dedup that section 9.5 requires.

func BumpSeen

func BumpSeen(q Q, peer, th, id string, now time.Time) error

BumpSeen raises the last seen id of a thread to id if it is larger.

func DeleteMirrored

func DeleteMirrored(q Q, sharer, th, of string) (int64, error)

DeleteMirrored forgets what sharer mirrored of its thread th with of.

func DropRef

func DropRef(q Q, peer, ref string) error

DropRef removes the queued chunks of a blob (the peer refused it).

func DropThread

func DropThread(q Q, peer, th string) error

DropThread removes everything queued for a thread.

func Enqueue

func Enqueue(q Q, r *OutboxRow) error

Enqueue appends a line to a peer's outbox.

func EnsurePeer

func EnsurePeer(q Q, key string, now time.Time) error

EnsurePeer creates a peer row if it does not exist.

func HasLog

func HasLog(q Q, peer, dir, id string) (bool, error)

HasLog reports whether a record exists.

func IsPrivate

func IsPrivate(q Q, peer, th string) (bool, error)

IsPrivate reports whether a thread is kept out of conversation sharing.

func PruneSeen

func PruneSeen(q Q, peer string, seen map[string]string, now time.Time) (int64, error)

PruneSeen drops outbox rows the peer's resume says it already has: in each listed thread, everything up to and including the seen id. The matching log rows are marked acked, since "seen" means durably received, and blobs whose chunks are all seen are sent.

func PutBlob

func PutBlob(q Q, b *Blob) error

PutBlob inserts or replaces a blob row.

func PutGrant

func PutGrant(q Q, g *GrantRow) error

PutGrant stores a grant under a role. Storing the same grant twice under the same role is a no-op.

func SetAcked

func SetAcked(q Q, peer, id string) error

SetAcked marks a sent message as acked.

func SetPrivate

func SetPrivate(q Q, peer, th string, now time.Time) (bool, error)

SetPrivate keeps a thread out of conversation sharing. It reports whether the thread was not private before.

func SetThreadState

func SetThreadState(q Q, peer, th string, mine bool, state, note string, now time.Time) error

SetThreadState records a state change, ours (mine) or the peer's. The side's since timestamp moves only when its state or note differs from what it was, so it says how long the side has been in that state.

func Terminal

func Terminal(state string) bool

Terminal reports whether a state ends a thread's active life.

func TouchPeer

func TouchPeer(q Q, key, name, about string, caps []string, now time.Time) error

TouchPeer records what a peer said about itself in hello.

func TouchThread

func TouchThread(q Q, peer, th, subject, origin string, now time.Time) error

TouchThread creates a thread on first sight (open on both sides since now), and otherwise bumps its update time and fills in a missing subject.

Types

type Addr

type Addr struct {
	Addr    string    `json:"addr"`
	Source  string    `json:"source"`
	Added   time.Time `json:"added"`
	LastOK  time.Time `json:"last_ok,omitzero"`
	LastErr string    `json:"last_err,omitempty"`
}

Addr is an address hint for a peer. Sources, from most to least authoritative: "connect" (given to us out of band), "introduce" and "hello" (the peer's own advertised address).

func Addrs

func Addrs(q Q, peer string) ([]Addr, error)

Addrs lists a peer's address hints, best first: the address that worked most recently, then by source.

type Blob

type Blob struct {
	Peer     string    `json:"peer"`
	Dir      string    `json:"dir"`
	Ref      string    `json:"ref"`
	Th       string    `json:"th,omitempty"`
	Name     string    `json:"name,omitempty"`
	Mime     string    `json:"mime,omitempty"`
	Size     int64     `json:"size"` // declared size, -1 if not yet known
	Received int64     `json:"received"`
	NextN    int       `json:"next_n"`
	Path     string    `json:"path,omitempty"`
	Status   string    `json:"status"` // in: pending; received (data in, awaiting the msg that names it); complete; refused. out: queued; sent (the msg that carries it was acked); refused
	Updated  time.Time `json:"updated"`
}

Blob tracks one blob transfer in either direction.

func GetBlob

func GetBlob(q Q, peer, dir, ref string) (*Blob, error)

GetBlob returns a blob or nil.

type Filter

type Filter struct {
	Peer       string
	Th         string
	Dirs       []string // default: all
	Types      []string // default: all
	AfterSeq   int64
	UnreadOnly bool
	Inbox      bool // received records plus local events worth a reader's attention
	Limit      int  // default 1000
	Last       bool // with Limit: the last N records, still returned in order
	// Mirrors includes mirrored lines (dir "mirror"), which are left out
	// unless asked for by this or by Dirs: they are other agents' threads,
	// not this agent's business.
	Mirrors bool
}

Filter selects log records.

type GrantQuery

type GrantQuery struct {
	Role     string
	Iss      string
	Sub      string
	Audience string
	ValidAt  time.Time // if set, only grants not expired at this time
}

GrantQuery selects grants. Empty fields match anything.

type GrantRow

type GrantRow struct {
	Hash     string    `json:"hash"`
	Role     string    `json:"role"`
	Iss      string    `json:"iss"`
	Sub      string    `json:"sub"`
	Caps     []string  `json:"caps"`
	Exp      time.Time `json:"exp"`
	Raw      []byte    `json:"-"`
	Audience string    `json:"audience,omitempty"`
	Created  time.Time `json:"created"`
}

GrantRow is a stored grant.

type MirroredThread

type MirroredThread struct {
	Sharer  string    `json:"sharer"` // the agent that shares it
	Of      string    `json:"of"`     // the other party
	Th      string    `json:"th"`
	Subject string    `json:"subject,omitempty"`
	Lines   int       `json:"lines"`
	Updated time.Time `json:"updated"`
}

MirroredThread is a thread another agent shares with this host.

type OutboxRow

type OutboxRow struct {
	Seq     int64
	Peer    string
	ID      string
	T       string
	Th      string
	Ref     string
	Line    []byte
	Created time.Time
}

OutboxRow is a sent line that the peer has not yet confirmed.

type Peer

type Peer struct {
	Key       string    `json:"key"`
	Name      string    `json:"name,omitempty"`
	About     string    `json:"about,omitempty"`
	Caps      []string  `json:"caps,omitempty"`
	Alias     string    `json:"alias,omitempty"`
	FirstSeen time.Time `json:"first_seen"`
	LastSeen  time.Time `json:"last_seen,omitzero"`
	Parked    bool      `json:"parked,omitempty"`
	Addrs     []Addr    `json:"addrs,omitempty"`
}

Peer is what we know about a remote key.

func GetPeer

func GetPeer(q Q, key string) (*Peer, error)

GetPeer returns a peer, or nil if unknown.

type PresenceRow

type PresenceRow struct {
	Origin   string
	Seq      int64
	Name     string
	TS       time.Time // the origin's timestamp
	Received time.Time
	Via      string // the peer it arrived from
	Hops     int
	Doc      []byte // exactly as signed
}

PresenceRow is the newest signed presence document from one origin.

type Q

type Q interface {
	Exec(query string, args ...any) (sql.Result, error)
	Query(query string, args ...any) (*sql.Rows, error)
	QueryRow(query string, args ...any) *sql.Row
}

Q is implemented by *sql.DB and *sql.Tx.

type Record

type Record struct {
	Seq   int64          `json:"seq"`
	Peer  string         `json:"peer"`
	Dir   string         `json:"dir"`
	ID    string         `json:"id"`
	T     string         `json:"t"`
	Th    string         `json:"th,omitempty"`
	Line  []byte         `json:"-"`
	At    time.Time      `json:"at"`
	Acked bool           `json:"acked,omitempty"`
	Read  bool           `json:"read,omitempty"`
	Meta  map[string]any `json:"meta,omitempty"`
}

Record is one entry in the message log. Dir is "in" (received), "out" (sent), "sys" (a local event such as a connection coming up) or "mirror" (a line of another agent's thread that it shares with this host; Peer is that agent, ID the original line's id, Meta says who the other party is).

type Store

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

Store is the database handle.

func Open

func Open(path string) (*Store, error)

Open opens (creating if needed) the database at path.

func (*Store) AddrResult

func (s *Store) AddrResult(peer, addr string, dialErr error, now time.Time) error

AddrResult records the outcome of dialing an address.

func (*Store) Blobs

func (s *Store) Blobs(peer string) ([]*Blob, error)

Blobs lists blobs for a peer ("" for all), newest first.

func (*Store) Close

func (s *Store) Close() error

Close closes the database.

func (*Store) DB

func (s *Store) DB() Q

DB returns the handle for reads outside a transaction.

func (*Store) DeleteGrant

func (s *Store) DeleteGrant(hash string) (int64, error)

DeleteGrant removes a grant in every role (local revocation).

func (*Store) ExpireOutbox

func (s *Store) ExpireOutbox(cutoff time.Time) (int64, error)

ExpireOutbox drops outbox rows created before cutoff (open question 2 proposes seven days) and returns how many went.

func (*Store) ExpirePresence

func (s *Store) ExpirePresence(cutoff time.Time) ([]*PresenceRow, error)

ExpirePresence deletes documents older than cutoff and returns them.

func (*Store) GetKV

func (s *Store) GetKV(k string) (string, bool, error)

GetKV returns a stored value.

func (*Store) Grants

func (s *Store) Grants(gq GrantQuery) ([]*GrantRow, error)

Grants returns matching grants, newest first.

func (*Store) MarkRead

func (s *Store) MarkRead(seqs []int64) error

MarkRead marks records as read.

func (*Store) MaxOwnID

func (s *Store) MaxOwnID() (string, error)

MaxOwnID returns the largest id this peer ever sent, so the id generator can keep increasing across restarts.

func (*Store) MaxSeq

func (s *Store) MaxSeq() (int64, error)

MaxSeq returns the newest log sequence number.

func (*Store) MirroredThreads

func (s *Store) MirroredThreads() ([]MirroredThread, error)

MirroredThreads lists the threads other agents share with this host.

func (*Store) OpenThreads

func (s *Store) OpenThreads(peer string, cutoff time.Time) (int, error)

OpenThreads counts a peer's threads that are neither done, failed nor closed on either side and were active since cutoff.

func (*Store) OutboxAfter

func (s *Store) OutboxAfter(peer string, after int64, limit int) ([]OutboxRow, error)

OutboxAfter returns up to limit outbox rows for peer with seq > after, in order.

func (*Store) OutboxCount

func (s *Store) OutboxCount(peer string) (int, error)

OutboxCount returns how many lines are waiting for peer ("" for all).

func (*Store) Peers

func (s *Store) Peers() ([]*Peer, error)

Peers lists all known peers, most recently seen first.

func (*Store) PeersWithOutbox

func (s *Store) PeersWithOutbox() ([]string, error)

PeersWithOutbox lists peers that have queued lines.

func (*Store) PresenceSince

func (s *Store) PresenceSince(cutoff time.Time) ([]*PresenceRow, error)

PresenceSince lists documents whose origin timestamp is at or after cutoff, most recent first.

func (*Store) PutPresence

func (s *Store) PutPresence(r *PresenceRow) (stored, isNew bool, err error)

PutPresence stores a document if it is newer (by seq) than what we have. It reports whether it was stored and whether the origin is new to us.

func (*Store) Query

func (s *Store) Query(f Filter) ([]Record, error)

Query returns log records matching f, in log order.

func (*Store) SeenMap

func (s *Store) SeenMap(peer string, cutoff time.Time) (map[string]string, error)

SeenMap returns the per-thread last seen ids for peer, for threads touched since cutoff (zero for all).

func (*Store) SetAlias

func (s *Store) SetAlias(key, alias string) error

SetAlias sets a local nickname for a peer.

func (*Store) SetKV

func (s *Store) SetKV(k, v string) error

SetKV stores a value.

func (*Store) SetMeta

func (s *Store) SetMeta(seq int64, add map[string]any) error

SetMeta merges keys into a record's metadata.

func (*Store) SetParked

func (s *Store) SetParked(key string, parked bool) error

SetParked marks a peer as parked (it said bye, or we did): no automatic reconnection until there is new outbound traffic.

func (*Store) Threads

func (s *Store) Threads(peer, th string) ([]*Thread, error)

Threads lists threads, most recently updated first. peer and th filter when non-empty.

func (*Store) TrimPresence

func (s *Store) TrimPresence(max int) error

TrimPresence keeps at most max origins, dropping the least recent. It bounds what a peer can make us store by inventing origins.

func (*Store) Tx

func (s *Store) Tx(fn func(q Q) error) error

Tx runs fn in a transaction. fn must use only the Q it is given: the store has a single connection, so touching s.DB() inside fn deadlocks.

type Thread

type Thread struct {
	Peer       string    `json:"peer"`
	Th         string    `json:"th"`
	Subject    string    `json:"subject,omitempty"`
	Origin     string    `json:"origin"`
	Created    time.Time `json:"created"`
	Updated    time.Time `json:"updated"`
	MyState    string    `json:"my_state"`
	MyNote     string    `json:"my_note,omitempty"`
	MySince    time.Time `json:"my_since"` // when my state or note last changed
	TheirState string    `json:"their_state"`
	TheirNote  string    `json:"their_note,omitempty"`
	TheirSince time.Time `json:"their_since"`
	Unread     int       `json:"unread,omitempty"`
}

Thread is a conversation with one peer.

func GetThread

func GetThread(q Q, peer, th string) (*Thread, error)

GetThread returns one thread or nil.

func (*Thread) Open

func (t *Thread) Open() bool

Open reports whether the thread still wants a live connection.

Jump to

Keyboard shortcuts

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