search

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Oct 10, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	CollectionLedgers         = "ledgers"
	CollectionBalances        = "balances"
	CollectionTransactions    = "transactions"
	CollectionReconciliations = "reconciliations"
	CollectionIdentities      = "identities"
)

Variables

View Source 
var (
	ErrReindexInProgress = errors.New("a reindex operation is already in progress")
	ErrReindexNotStarted = errors.New("no reindex operation has been started")
	ErrReindexClosed     = errors.New("reindex service is closed")
)
View Source 
var ErrMemoryBackpressure = errors.New("typesense memory backpressure")

ErrMemoryBackpressure marks an indexing task that Typesense rejected at its native memory limit. Workers use this sentinel to retry without consuming their normal failure budget.

Functions

func IsMemoryBackpressure 

func IsMemoryBackpressure(err error) bool

IsMemoryBackpressure reports whether Typesense rejected a write because its native memory-used-max-percentage limit was reached.

func MarkMemoryBackpressure 

func MarkMemoryBackpressure(err error) error

MarkMemoryBackpressure wraps a native Typesense rejection with the sentinel used by the index worker's retry policy.

func NormalizeTransactionDocument 

func NormalizeTransactionDocument(data map[string]interface{}) map[string]interface{}

NormalizeTransactionDocument applies the same field normalization as a transactions upsert (metadata coercion, bigint strings, required defaults, time fields) without writing to Typesense. Used by tests to verify reindex payloads retain indexed fields.

func TryReindexIfNeeded 

func TryReindexIfNeeded(ctx context.Context, client *TypesenseClient, ds ReindexRepository)

Types

type CollectionConfig 

type CollectionConfig struct {
	Schema           *api.CollectionSchema
	IDField          string
	TimeFields       []string
	BigIntFields     []string
	DefaultSortField string
}

CollectionConfig holds configuration for a specific collection.

type IndexBatch 

type IndexBatch struct {
	ID           string      `json:"id"`
	Dependencies []IndexItem `json:"dependencies"` // Items that must be indexed first (e.g., balances)
	Primary      *IndexItem  `json:"primary"`      // Primary item indexed after dependencies (e.g., transaction)
	CreatedAt    time.Time   `json:"created_at"`
}

IndexBatch represents a batch of items to be indexed in dependency order. Dependencies are indexed first, then the primary item is indexed. This ensures referential integrity in the search index.

func NewIndexBatch 

func NewIndexBatch(id string) *IndexBatch

NewIndexBatch creates a new IndexBatch with the given ID.

func (*IndexBatch) AddDependency 

func (b *IndexBatch) AddDependency(collection, documentID string, data interface{})

AddDependency adds a dependency item to the batch. Dependencies are indexed before the primary item.

func (*IndexBatch) Deduplicate 

func (b *IndexBatch) Deduplicate()

Deduplicate removes duplicate dependencies based on collection and document ID.

func (*IndexBatch) SetPrimary 

func (b *IndexBatch) SetPrimary(collection, documentID string, data interface{})

SetPrimary sets the primary item to be indexed after all dependencies.

type IndexItem 

type IndexItem struct {
	Collection string      `json:"collection"` // Collection name (e.g., "balances", "transactions")
	DocumentID string      `json:"document_id"`
	Data       interface{} `json:"data"`
}

IndexItem represents a single item to be indexed in Typesense.

type MultiSearchRequest 

type MultiSearchRequest struct {
	Searches []api.MultiSearchSearchesParameter `json:"searches"`
}

MultiSearchRequest represents a request for multiple searches

type NotificationPayload 

type NotificationPayload struct {
	Table string                 `json:"table"`
	Data  map[string]interface{} `json:"data"`
}

NotificationPayload represents the payload structure for notifications, containing the table and data.

type ReindexConfig 

type ReindexConfig struct {
	BatchSize                 int
	BackpressureRetryInterval time.Duration
}

ReindexConfig holds configuration for reindexing.

type ReindexJobs 

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

ReindexJobs coordinates background jobs for one ledger instance.

func NewReindexJobs 

func NewReindexJobs(client *TypesenseClient, repository ReindexRepository, retryInterval time.Duration) *ReindexJobs

func (*ReindexJobs) Close 

func (j *ReindexJobs) Close()

func (*ReindexJobs) Progress 

func (j *ReindexJobs) Progress() (ReindexProgress, error)

func (*ReindexJobs) Start 

func (j *ReindexJobs) Start(batchSize int) (ReindexProgress, error)

func (*ReindexJobs) TryReindexIfNeeded 

func (j *ReindexJobs) TryReindexIfNeeded(ctx context.Context)

TryReindexIfNeeded uses the same admission and progress state as manual jobs.

type ReindexProgress 

type ReindexProgress struct {
	Status           string     `json:"status"` // "in_progress", "completed", "failed"
	Phase            string     `json:"phase"`  // "drop_collections", "indexing_ledgers", etc.
	TotalRecords     int64      `json:"total_records"`
	ProcessedRecords int64      `json:"processed_records"`
	Errors           []string   `json:"errors,omitempty"`
	StartedAt        time.Time  `json:"started_at"`
	CompletedAt      *time.Time `json:"completed_at,omitempty"`
	Paused           bool       `json:"paused"`
	PauseReason      string     `json:"pause_reason,omitempty"`
	PausedAt         *time.Time `json:"paused_at,omitempty"`
}

ReindexProgress tracks the progress of a reindex operation.

type ReindexRepository 

type ReindexRepository interface {
	GetAllLedgers(int, int) ([]model.Ledger, error)
	GetAllIdentitiesPaginated(int, int) ([]model.Identity, error)
	GetAllBalances(int, int) ([]model.Balance, error)
	GetAllTransactions(context.Context, int, int) ([]model.Transaction, error)
}

ReindexRepository supplies the projections needed to rebuild the search index.

type ReindexService 

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

ReindexService handles reindexing operations.

func NewReindexService 

func NewReindexService(client *TypesenseClient, datasource ReindexRepository, config ReindexConfig) *ReindexService

NewReindexService creates a new ReindexService instance.

func (*ReindexService) GetProgress 

func (r *ReindexService) GetProgress() ReindexProgress

GetProgress returns the current progress of the reindex operation.

func (*ReindexService) GetProgressPtr 

func (r *ReindexService) GetProgressPtr() *ReindexProgress

GetProgressPtr returns a pointer to the current progress.

func (*ReindexService) StartReindex 

func (r *ReindexService) StartReindex(ctx context.Context) (*ReindexProgress, error)

StartReindex performs a complete reindex of all data. It drops all collections, recreates them, and indexes data in order: ledgers -> identities -> balances -> transactions

type TypesenseClient 

type TypesenseClient struct {
	Client *typesense.Client
}

TypesenseClient wraps the Typesense client and provides methods to interact with it.

func NewTypesenseClient 

func NewTypesenseClient(apiKey string, hosts []string) *TypesenseClient

NewTypesenseClient initializes and returns a new Typesense client instance.

func (*TypesenseClient) CreateCollection 

func (t *TypesenseClient) CreateCollection(ctx context.Context, schema *api.CollectionSchema) (*api.CollectionResponse, error)

CreateCollection creates a collection in Typesense based on the provided schema. If the collection already exists, it will return without error.

func (*TypesenseClient) DropAllCollections 

func (t *TypesenseClient) DropAllCollections(ctx context.Context) error

DropAllCollections drops all known collections from Typesense.

func (*TypesenseClient) DropCollection 

func (t *TypesenseClient) DropCollection(ctx context.Context, collectionName string) error

DropCollection deletes a collection from Typesense. A missing collection is not an error; Typesense reports it variously as "not found", "Not Found", or a 404 with "No collection with name X found." depending on version.

func (*TypesenseClient) EnsureCollectionsExist 

func (t *TypesenseClient) EnsureCollectionsExist(ctx context.Context) error

EnsureCollectionsExist ensures that all the necessary collections exist in the Typesense schema. If a collection doesn't exist, it will create the collection based on the latest schema.

func (*TypesenseClient) HandleBatchNotification 

func (t *TypesenseClient) HandleBatchNotification(ctx context.Context, batch *IndexBatch) error

HandleBatchNotification processes a batch of items and indexes them in dependency order. It first indexes all dependencies (e.g., balances), then indexes the primary item (e.g., transaction). This ensures referential integrity in the search index.

func (*TypesenseClient) HandleNotification 

func (t *TypesenseClient) HandleNotification(ctx context.Context, table string, data map[string]interface{}) error

HandleNotification processes incoming notifications and updates Typesense collections based on the table and data. It ensures the required fields exist and upserts the data into Typesense.

func (*TypesenseClient) MigrateTypeSenseSchema 

func (t *TypesenseClient) MigrateTypeSenseSchema(ctx context.Context, collectionName string) error

MigrateTypeSenseSchema adds new fields and removes old fields from the existing collection schema in Typesense. This is useful when the schema has been updated, and fields need to be added or removed.

func (*TypesenseClient) MultiSearch 

func (t *TypesenseClient) MultiSearch(ctx context.Context, searchRequests api.MultiSearchSearchesParameter) (*api.MultiSearchResult, error)

Remove the incorrect MultiSearch method and replace with this:

func (*TypesenseClient) Search 

func (t *TypesenseClient) Search(ctx context.Context, collection string, searchParams *api.SearchCollectionParams) (*api.SearchResult, error)

Search performs a search query on a specific collection with the provided search parameters.

Source Files

View all Source files

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.