store

package
v0.1.22 Latest Latest
Warning

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

Go to latest
Published: Sep 2, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package store persists the message log and the PNR state.

The two are deliberately separate concerns with different guarantees.

The message log is append-only and holds the exact bytes received or sent. It is written before anything is interpreted, so a parser bug costs a reprocessing run rather than a lost booking, and so "what did we actually receive at 14:32" has an answer that does not depend on the parser that was deployed at the time.

PNR state is derived. Every change is recorded as an event carrying the id of the message that caused it, and the current record is a projection of those events. That is what makes an interline dispute answerable: the record shows what it holds now, and the events show which partner message put it there.

Index

Constants

View Source
const (
	// QueueConfirmation holds records a partner has confirmed, where the
	// holding side has to be told or the itinerary reissued.
	QueueConfirmation = "confirmation"
	// QueueUnable holds records a partner refused. Someone has to rebook.
	QueueUnable = "unable"
	// QueueWaitlist holds records with waitlisted segments awaiting clearance.
	QueueWaitlist = "waitlist"
	// QueuePending holds requests a partner has not answered inside the agreed
	// time. Nothing is wrong with the record; the conversation has stalled.
	QueuePending = "pending"
	// QueueTicketing holds records whose ticketing time limit is near or past.
	QueueTicketing = "ticketing"
	// QueueScheduleChange holds records touched by a schedule message: the
	// flight they are booked on has moved, changed equipment or been
	// cancelled, and the passenger has not been told.
	QueueScheduleChange = "schedule-change"
	// QueueDivergence holds records where our state and a partner's disagree,
	// which is the case a human must always see.
	QueueDivergence = "divergence"
	// QueueGeneral is the fallback for placements with no better home.
	QueueGeneral = "general"
)

A queue is a list of records waiting for someone to do something about them.

It is the mechanism a reservations system uses to turn an asynchronous partner conversation into work: a carrier confirms a segment overnight, a ticketing deadline passes, a request goes unanswered, and each of those has to surface somewhere a person or a robot will look. Without queues those events are visible only to whoever happens to re-read the record.

Queue names here are Jetway's own vocabulary, not an IATA standard. Real systems number their queues and the numbering is house-specific, so a deployment that has to match an existing convention should map these names onto its own numbers at the edge rather than expect the numbers to be meaningful across systems.

Variables

View Source
var (
	// ErrNotFound is returned when a record or message does not exist.
	ErrNotFound = errors.New("store: not found")
	// ErrConflict is returned when an update's expected version does not match
	// the stored version. The caller must re-read and retry: a blind overwrite
	// loses whichever concurrent change it did not see.
	ErrConflict = errors.New("store: version conflict")
	// ErrDuplicate is returned when a record locator is already taken.
	ErrDuplicate = errors.New("store: duplicate")
)

Errors returned by every implementation.

Queues lists the known queue names in the order a console should show them.

Functions

func MigrateSchema

func MigrateSchema(ctx context.Context, s *Postgres) error

MigrateSchema applies any migrations the database has not seen.

Each migration runs inside its own transaction together with the row that records it, so a failure part-way leaves the database at a version that actually reflects its contents. Applying is idempotent, which is what makes running it on every start safe and removes a manual deployment step.

func Migrations

func Migrations() ([]migration, error)

Migrations returns the embedded schema changes in version order.

func NormaliseFlightKey

func NormaliseFlightKey(key string) string

NormaliseFlightKey renders a carrier and flight number in the one spelling the lookups compare against.

Carriers write the same flight both zero-padded and bare, sometimes in the same conversation, so a key is only useful if both spellings collapse onto it. The designator is taken as the first two characters, never by scanning for the first digit: IATA designators are two characters and a third of them are alphanumeric -- U2, 4U, 2B -- so "the number starts at the first digit" split easyJet's own designator in half and no U2 flight could ever be matched to a schedule change. Found by the world simulator, whose carriers come from the real registry; every hand-written test here had used BA.

func SchemaSQL

func SchemaSQL() (string, error)

SchemaSQL returns every migration concatenated, for inspection and for bootstrapping a database by hand.

func SegmentOnFlight

func SegmentOnFlight(seg *pnr.Segment, flightKey, wireDate string) bool

SegmentOnFlight reports whether a segment is a live holding on a flight.

Both backends decide with this function rather than each with its own predicate. The Postgres lookup narrows with an index and then asks this, so an index that over-matches -- and containment does, because it will match a cancelled segment as happily as a live one -- cannot change the answer.

Types

type Diagnostic

type Diagnostic struct {
	Layer    string `json:"layer"` // "typeb", "edifact", "airimp", "padis"
	Severity string `json:"severity"`
	Code     string `json:"code"`
	Detail   string `json:"detail"`
	Line     int    `json:"line,omitempty"`
}

Diagnostic is a decoder observation, flattened across codec packages so the console and the API present one shape.

type Direction

type Direction string

Direction distinguishes traffic we received from traffic we sent.

const (
	Inbound  Direction = "in"
	Outbound Direction = "out"
)

type Event

type Event struct {
	ID        string          `json:"id"`
	PNRID     string          `json:"pnr_id"`
	Seq       int64           `json:"seq"`
	Type      string          `json:"type"`
	Detail    string          `json:"detail"`
	Payload   json.RawMessage `json:"payload,omitempty"`
	MessageID string          `json:"message_id,omitempty"`
	Actor     string          `json:"actor,omitempty"`
	At        time.Time       `json:"at"`
}

Event is one applied change to a record.

type Format

type Format string

Format names the wire encoding.

const (
	FormatTypeB   Format = "typeb"
	FormatEDIFACT Format = "edifact"
	FormatUnknown Format = "unknown"
)

type Lookup

type Lookup interface {
	// FindPNRByDocument returns the record holding a document, by its compact
	// thirteen-digit number. Not found is (nil, nil): a document this node
	// never issued is an ordinary answer, not a failure.
	FindPNRByDocument(ctx context.Context, compactNumber string) (*pnr.PNR, error)

	// FindPNRByExternalLocator returns the record carrying another system's
	// locator. Owner may be empty to match any.
	FindPNRByExternalLocator(ctx context.Context, owner, value string) (*pnr.PNR, error)

	// FindPNRsByFlight returns every live record holding a segment on a
	// flight. wireDate may be empty to match every date, which is what a
	// schedule message covering a period needs.
	//
	// flightKey is the carrier and flight number with leading zeros removed,
	// because carriers write the same flight both ways and a schedule change
	// that misses half its holdings is worse than useless.
	FindPNRsByFlight(ctx context.Context, flightKey, wireDate string, limit int) ([]*pnr.PNR, error)

	// FindPNRsStale returns live records untouched since before the given
	// time, most overdue first.
	//
	// The ordering is the point. The sweeper used to read the most recently
	// updated records and look for stale ones among them, which is inverted:
	// the freshest records are precisely not the stale ones. Ordering by the
	// thing that makes a record due means a limit drops the least urgent work
	// rather than all of it.
	FindPNRsStale(ctx context.Context, before time.Time, limit int) ([]*pnr.PNR, error)

	// FindPNRsDueBy returns live records owing a ticketing time limit before
	// the given time, soonest deadline first.
	FindPNRsDueBy(ctx context.Context, deadline time.Time, limit int) ([]*pnr.PNR, error)
}

Lookups that find a record by something other than its own locator.

These exist because the alternative was a bug rather than merely a slow query. Three places on the hot path used to walk ListPNRs, which is "ORDER BY updated_at DESC LIMIT n" -- so they searched only recently touched records and reported *not found* for anything older. A ticket control message about a booking made last month was refused with "no record holds this document", which is not slow, it is false, and it gets more likely as the store grows.

So the contract here is total: an implementation must search every record or return an error. It must never quietly answer from a prefix.

type Mem

type Mem struct {
	// MaxMessages and MaxRecords bound what is retained, oldest discarded
	// first. Zero means unbounded, which is right for a test and wrong for
	// anything reachable from the internet: without a bound, a public demo is
	// a memory leak with a submit button.
	MaxMessages int
	MaxRecords  int

	// Now, when set, stamps defaults instead of the wall clock. Set before
	// use; read without a lock.
	Now func() time.Time
	// contains filtered or unexported fields
}

Mem is an in-memory Store.

It exists so the gateway runs with no external dependency -- for the demo, for tests, and for a carrier evaluating the message flow before provisioning a database. It is not a production backend: nothing survives a restart.

func NewMem

func NewMem() *Mem

NewMem returns an empty in-memory store.

func (*Mem) AppendMessage

func (s *Mem) AppendMessage(ctx context.Context, m *Message) error

func (*Mem) Close

func (s *Mem) Close() error

func (*Mem) CreatePNR

func (s *Mem) CreatePNR(ctx context.Context, p *pnr.PNR, events []Event) error

func (*Mem) DividePNR

func (s *Mem) DividePNR(ctx context.Context, parent *pnr.PNR, expected int64,
	child *pnr.PNR, parentEvents, childEvents []Event) error

func (*Mem) Enqueue

func (s *Mem) Enqueue(ctx context.Context, item *QueueItem) error

func (*Mem) Events

func (s *Mem) Events(ctx context.Context, pnrID string) ([]Event, error)

func (*Mem) FindByDedupKey

func (s *Mem) FindByDedupKey(ctx context.Context, peer, key string) (string, bool, error)

func (*Mem) FindOutboundByKey

func (s *Mem) FindOutboundByKey(ctx context.Context, peer, key string) (string, bool, error)

func (*Mem) FindPNRByDocument

func (s *Mem) FindPNRByDocument(ctx context.Context, compactNumber string) (*pnr.PNR, error)

func (*Mem) FindPNRByExternalLocator

func (s *Mem) FindPNRByExternalLocator(ctx context.Context, owner, value string) (*pnr.PNR, error)

func (*Mem) FindPNRsByFlight

func (s *Mem) FindPNRsByFlight(ctx context.Context, flightKey, wireDate string, limit int) ([]*pnr.PNR, error)

func (*Mem) FindPNRsDueBy

func (s *Mem) FindPNRsDueBy(ctx context.Context, deadline time.Time, limit int) ([]*pnr.PNR, error)

func (*Mem) FindPNRsStale

func (s *Mem) FindPNRsStale(ctx context.Context, before time.Time, limit int) ([]*pnr.PNR, error)

func (*Mem) GetMessage

func (s *Mem) GetMessage(ctx context.Context, id string) (*Message, error)

func (*Mem) GetPNR

func (s *Mem) GetPNR(ctx context.Context, locator string) (*pnr.PNR, error)

func (*Mem) GetPNRByID

func (s *Mem) GetPNRByID(ctx context.Context, id string) (*pnr.PNR, error)

func (*Mem) ListMessages

func (s *Mem) ListMessages(ctx context.Context, f MessageFilter) ([]*Message, error)

func (*Mem) ListPNRs

func (s *Mem) ListPNRs(ctx context.Context, limit int) ([]*pnr.PNR, error)

func (*Mem) ListQueue

func (s *Mem) ListQueue(ctx context.Context, f QueueFilter) ([]*QueueItem, error)

func (*Mem) NextLocatorCounter

func (s *Mem) NextLocatorCounter(ctx context.Context) (uint64, error)

func (*Mem) QueueCounts

func (s *Mem) QueueCounts(ctx context.Context) (map[string]int, error)

func (*Mem) UpdateMessage

func (s *Mem) UpdateMessage(ctx context.Context, m *Message) error

func (*Mem) UpdatePNR

func (s *Mem) UpdatePNR(ctx context.Context, p *pnr.PNR, expected int64, events []Event) error

func (*Mem) WorkQueueItem

func (s *Mem) WorkQueueItem(ctx context.Context, id, by, note string) error

type Message

type Message struct {
	ID        string    `json:"id"`
	Direction Direction `json:"direction"`
	At        time.Time `json:"at"`

	// Transport and Peer say how it arrived and from whom. Peer is the link
	// name, which is what routing and per-partner policy key off.
	Transport string `json:"transport"`
	Peer      string `json:"peer"`

	Format Format `json:"format"`
	// Kind is the decoded message type, e.g. "AIRIMP/sell" or "PAORES".
	Kind string `json:"kind,omitempty"`

	// Raw is the exact bytes on the wire. Never regenerate it from a parse.
	Raw []byte `json:"-"`
	// SHA256 is the hex digest of Raw, used for content-based deduplication
	// where an application-level reference is unavailable.
	SHA256 string `json:"sha256"`
	Size   int    `json:"size"`

	Status Status `json:"status"`
	Error  string `json:"error,omitempty"`

	// DedupKey is the application-level idempotency key: an EDIFACT interchange
	// control reference, or a Type B origin and time group. Empty when the
	// message class carries none.
	DedupKey string `json:"dedup_key,omitempty"`

	// TraceID and SpanID tie this message to the trace that handled it. They
	// turn "what happened to this message" from a search into a link, and they
	// survive in the log after the trace itself has been sampled away.
	TraceID string `json:"trace_id,omitempty"`
	SpanID  string `json:"span_id,omitempty"`

	// PossibleDuplicate records the Type B PDM indicator. Inbound it means the
	// sender said this may be a retransmission; outbound it means we marked it
	// as one on redelivery. A duplicate that arrives flagged is the protocol
	// working; one that arrives unflagged is worth an operator's attention.
	PossibleDuplicate bool `json:"possible_duplicate,omitempty"`

	// PNRID links the message to the record it touched.
	PNRID string `json:"pnr_id,omitempty"`
	// CorrelationID ties a response back to the request that provoked it, and
	// is what turns a message list into a conversation.
	CorrelationID string `json:"correlation_id,omitempty"`

	Diagnostics []Diagnostic `json:"diagnostics,omitempty"`
}

Message is one unit of traffic, inbound or outbound.

func (*Message) RawString

func (m *Message) RawString() string

RawString returns the message body as text, for display.

type MessageFilter

type MessageFilter struct {
	Peer   string
	PNRID  string
	Status Status
	Limit  int
	// SinceID returns only messages with an id greater than this one, which is
	// how the console tails the log without re-fetching.
	SinceID string
}

MessageFilter narrows a message listing.

type Postgres

type Postgres struct {

	// Now, when set, stamps defaults instead of the wall clock. Set before
	// use; read without a lock.
	Now func() time.Time
	// contains filtered or unexported fields
}

Postgres is the production Store.

func OpenPostgres

func OpenPostgres(ctx context.Context, dsn string) (*Postgres, error)

OpenPostgres connects and verifies the schema is present.

func (*Postgres) AppendMessage

func (s *Postgres) AppendMessage(ctx context.Context, m *Message) error

func (*Postgres) Close

func (s *Postgres) Close() error

func (*Postgres) CreatePNR

func (s *Postgres) CreatePNR(ctx context.Context, p *pnr.PNR, events []Event) error

func (*Postgres) DividePNR

func (s *Postgres) DividePNR(ctx context.Context, parent *pnr.PNR, expected int64,
	child *pnr.PNR, parentEvents, childEvents []Event) error

func (*Postgres) Enqueue

func (s *Postgres) Enqueue(ctx context.Context, item *QueueItem) error

func (*Postgres) Events

func (s *Postgres) Events(ctx context.Context, pnrID string) ([]Event, error)

func (*Postgres) FindByDedupKey

func (s *Postgres) FindByDedupKey(ctx context.Context, peer, key string) (string, bool, error)

func (*Postgres) FindOutboundByKey

func (s *Postgres) FindOutboundByKey(ctx context.Context, peer, key string) (string, bool, error)

func (*Postgres) FindPNRByDocument

func (s *Postgres) FindPNRByDocument(ctx context.Context, compactNumber string) (*pnr.PNR, error)

func (*Postgres) FindPNRByExternalLocator

func (s *Postgres) FindPNRByExternalLocator(ctx context.Context, owner, value string) (*pnr.PNR, error)

func (*Postgres) FindPNRsByFlight

func (s *Postgres) FindPNRsByFlight(ctx context.Context, flightKey, wireDate string, limit int) ([]*pnr.PNR, error)

func (*Postgres) FindPNRsDueBy

func (s *Postgres) FindPNRsDueBy(ctx context.Context, deadline time.Time, limit int) ([]*pnr.PNR, error)

func (*Postgres) FindPNRsStale

func (s *Postgres) FindPNRsStale(ctx context.Context, before time.Time, limit int) ([]*pnr.PNR, error)

func (*Postgres) GetMessage

func (s *Postgres) GetMessage(ctx context.Context, id string) (*Message, error)

func (*Postgres) GetPNR

func (s *Postgres) GetPNR(ctx context.Context, locator string) (*pnr.PNR, error)

func (*Postgres) GetPNRByID

func (s *Postgres) GetPNRByID(ctx context.Context, id string) (*pnr.PNR, error)

func (*Postgres) ListMessages

func (s *Postgres) ListMessages(ctx context.Context, f MessageFilter) ([]*Message, error)

func (*Postgres) ListPNRs

func (s *Postgres) ListPNRs(ctx context.Context, limit int) ([]*pnr.PNR, error)

func (*Postgres) ListQueue

func (s *Postgres) ListQueue(ctx context.Context, f QueueFilter) ([]*QueueItem, error)

func (*Postgres) NextLocatorCounter

func (s *Postgres) NextLocatorCounter(ctx context.Context) (uint64, error)

func (*Postgres) QueueCounts

func (s *Postgres) QueueCounts(ctx context.Context) (map[string]int, error)

func (*Postgres) UpdateMessage

func (s *Postgres) UpdateMessage(ctx context.Context, m *Message) error

func (*Postgres) UpdatePNR

func (s *Postgres) UpdatePNR(ctx context.Context, p *pnr.PNR, expected int64, events []Event) error

func (*Postgres) WorkQueueItem

func (s *Postgres) WorkQueueItem(ctx context.Context, id, by, note string) error

type QueueFilter

type QueueFilter struct {
	// Queue restricts to one queue name. Empty means every queue.
	Queue string
	// PNRID restricts to one record.
	PNRID string
	// IncludeWorked returns cleared items as well as pending ones. The default
	// is the working view: what still needs doing.
	IncludeWorked bool
	Limit         int
}

QueueFilter narrows a queue listing.

type QueueItem

type QueueItem struct {
	ID    string `json:"id"`
	Queue string `json:"queue"`

	PNRID   string `json:"pnr_id"`
	Locator string `json:"locator,omitempty"`

	// Code is the stable machine-readable reason, e.g. "tktl_expired". Together
	// with Queue, PNRID and SegmentRef it is the idempotency key: the same
	// segment is on a queue once per reason until the item is worked, so a
	// sweeper may run as often as it likes.
	Code string `json:"code"`
	// Reason is the human-readable form of the same thing.
	Reason string `json:"reason"`

	// MessageID is the message that caused the placement, where there was one.
	MessageID string `json:"message_id,omitempty"`
	// SegmentRef narrows the placement to one segment, where it applies.
	SegmentRef int `json:"segment_ref,omitempty"`

	PlacedAt time.Time `json:"placed_at"`
	PlacedBy string    `json:"placed_by"`

	// WorkedAt is nil while the item is pending.
	WorkedAt *time.Time `json:"worked_at,omitempty"`
	WorkedBy string     `json:"worked_by,omitempty"`
	// Note is what the worker said when clearing it.
	Note string `json:"note,omitempty"`
}

QueueItem is one record placed on one queue for one reason.

An item is evidence as much as it is a task: it says what put the record here, which message caused it, and when. Working an item does not delete it, because "who cleared this and when" is exactly the question asked after an interline dispute.

func (*QueueItem) Pending

func (q *QueueItem) Pending() bool

Pending reports whether the item is still outstanding.

type QueueStore

type QueueStore interface {
	// Enqueue places a record on a queue. It returns ErrDuplicate when the same
	// segment of the same record is already pending on the same queue for the
	// same code, which is what makes a repeatedly-running sweeper harmless.
	Enqueue(ctx context.Context, item *QueueItem) error
	// WorkQueueItem marks an item done. Working an already-worked item returns
	// ErrConflict rather than silently overwriting who cleared it.
	WorkQueueItem(ctx context.Context, id, by, note string) error
	// ListQueue returns items newest first.
	ListQueue(ctx context.Context, f QueueFilter) ([]*QueueItem, error)
	// QueueCounts returns the number of pending items per queue name.
	QueueCounts(ctx context.Context) (map[string]int, error)
}

QueueStore is the queue half of the persistence contract. It is separate from Store only for readability; every backend implements both.

type Status

type Status string

Status tracks a message through the pipeline.

const (
	// StatusReceived means the bytes are durable but nothing has read them.
	StatusReceived Status = "received"
	// StatusDecoded means the envelope and body parsed.
	StatusDecoded Status = "decoded"
	// StatusApplied means the message changed a record, or was correctly
	// determined to require no change.
	StatusApplied Status = "applied"
	// StatusRejected means the message was understood and refused, for example
	// as a duplicate or as test traffic on a production link.
	StatusRejected Status = "rejected"
	// StatusDLQ means the pipeline could not process it and a human must look.
	// Messages never leave the system on this path; they wait to be replayed.
	StatusDLQ Status = "dlq"
	// StatusSent applies to outbound traffic handed to a transport.
	StatusSent Status = "sent"
	// StatusUndeliverable applies to outbound traffic no transport accepted.
	StatusUndeliverable Status = "undeliverable"
	// StatusAcknowledged means a partner confirmed receipt of outbound traffic,
	// which on an EDIFACT link is a CONTRL saying so. Delivery and
	// acknowledgement are different facts: a transport that took the bytes
	// proves nothing about whether the partner could read them.
	StatusAcknowledged Status = "acknowledged"
	// StatusRefused means a partner received outbound traffic and rejected it.
	StatusRefused Status = "refused"
)

type Store

type Store interface {
	// AppendMessage writes a message. The raw bytes must be durable before it
	// returns, because the caller acknowledges the peer on success.
	AppendMessage(ctx context.Context, m *Message) error
	// UpdateMessage records a status transition and any diagnostics.
	UpdateMessage(ctx context.Context, m *Message) error
	GetMessage(ctx context.Context, id string) (*Message, error)
	ListMessages(ctx context.Context, f MessageFilter) ([]*Message, error)

	// FindByDedupKey returns the id of an earlier inbound message from the same
	// peer with the same application-level key, which is how a retransmission
	// is recognised without re-applying it.
	FindByDedupKey(ctx context.Context, peer, key string) (string, bool, error)

	// FindOutboundByKey returns the id of a message sent to a peer carrying the
	// given application-level key. It is the other direction of the same
	// question, and is how an acknowledgement is matched to what it
	// acknowledges.
	FindOutboundByKey(ctx context.Context, peer, key string) (string, bool, error)

	// CreatePNR stores a new record at version 1 along with the events that
	// created it.
	CreatePNR(ctx context.Context, p *pnr.PNR, events []Event) error
	// UpdatePNR stores a record, failing with ErrConflict unless the stored
	// version still equals expectedVersion.
	UpdatePNR(ctx context.Context, p *pnr.PNR, expectedVersion int64, events []Event) error
	GetPNR(ctx context.Context, locator string) (*pnr.PNR, error)
	GetPNRByID(ctx context.Context, id string) (*pnr.PNR, error)
	ListPNRs(ctx context.Context, limit int) ([]*pnr.PNR, error)
	Events(ctx context.Context, pnrID string) ([]Event, error)

	// NextLocatorCounter returns a value that has never been returned before.
	// It is the uniqueness source behind record locator allocation.
	NextLocatorCounter(ctx context.Context) (uint64, error)

	// Lookup finds records by document, external locator or flight.
	// DividePNR writes a division: the parent as it now stands and the new
	// child, atomically, under the parent's expected version.
	//
	// It is one method rather than a create followed by an update because a
	// division is one change to two records. Done as two writes, a version
	// conflict on the second leaves the child created and the parent not
	// updated -- both records then list the same passengers, which is a torn
	// booking that no partner can be told about coherently. Under concurrency
	// that is not a rare case: a carrier reply landing between the read and
	// the write is exactly what optimistic concurrency is there to catch.
	DividePNR(ctx context.Context, parent *pnr.PNR, expected int64, child *pnr.PNR,
		parentEvents, childEvents []Event) error

	Lookup

	// QueueStore holds the work queues records are placed on.
	QueueStore

	Close() error
}

Store is the persistence contract.

Implementations must be safe for concurrent use: a gateway processes many links at once, and two of them can touch the same record.

Jump to

Keyboard shortcuts

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