database

package
v0.8.77 Latest Latest
Warning

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

 Go to latest Published: Oct 18, 2026 License: AGPL-3.0, AGPL-3.0-or-later, Apache-2.0 Imports: 27 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	AliasesRepoName = "/ALIASES"

	// MaxAliases is how many devices may stay paired to a node at once.
	MaxAliases = 10
)
View Source 
const (
	AuthRepoName    = "/AUTH"
	PassSubName     = "PASS" // TODO pass restore functionality
	DefaultOwnerKey = "OWNER"
)
View Source 
const (
	ChatNamespace     = "/CHATS"
	MessageNamespace  = "/MESSAGES"
	NonceSubNamespace = "NONCE"
)
View Source 
const (
	MediaRepoName          = "/MEDIA"
	ChatMediaRepoName      = "/CHATMEDIA"
	SponsoredMediaRepoName = "/SPONSOREDMEDIA"
	ImageSubNamespace      = "IMAGES"
	VideoSubNamespace      = "VIDEOS"
	CopySubNamespace       = "COPIES"
)
View Source 
const (
	BlocklistSubNamespace     = "BLOCKLIST"
	BlocklistUserSubNamespace = "USER"
	BlocklistTermSubNamespace = "TERM"
)
View Source 
const (
	PollRepoName = "/POLLS"

	// Poll-owned key segments. They deliberately duplicate the values likes
	// use rather than borrowing ReactionRepoName's constants: the two keyspaces
	// are independent, and sharing a constant would silently move the poll
	// keys if likes ever renamed theirs.
	VotesSubNamespace     = "VOTES" // per-option vote counters
	VotesIncrSubNamespace = "INCR"
	VotesDecrSubNamespace = "DECR"
	VoterSubNamespace     = "VOTER" // per-voter record of the option they picked
)
View Source 
const (
	ReactionRepoName    = "/REACTIONS"
	IncrSubNamespace    = "INCR"
	ReactorSubNamespace = "REACTOR"
	ReactedSubNamespace = "REACTED" // per-user index of reacted tweet refs
	EmojiSubNamespace   = "EMOJI"   // per-emoji counters, one key per reaction

)
View Source 
const (
	UsersRepoName = "/USERS"

	DefaultWarpnetUserNetwork = "warpnet"
)
View Source 
const (
	BlocksRepoName = "/BLOCKS"
)
View Source 
const BookmarkRepoName = "/BOOKMARKS"
View Source 
const (
	ErrNilNodeRepo = local_store.DBError("node repo is nil")
)

slash is required because of: invalid datastore key: NODES:/peers/keys/AASAQAISEAXNRKHMX2O3AA26JM7NGIWUPOGIITJ2UHHXGX4OWIEKPNAW6YCSK/priv

View Source 
const FilterRepoName = "/FILTERS"
View Source 
const (
	FollowRepoName = "/FOLLOW"
)
View Source 
const (
	MutesRepoName = "/MUTES"
)
View Source 
const (
	NotificationsRepoName = "/NOTIFICATIONS"
)
View Source 
const OrderRepoName = "/ORDERS"
View Source 
const OutboxNamespace = "/OUTBOX"
View Source 
const SettingsRepoName = "/SETTINGS"
View Source 
const (
	// SubscriptionsRepoName holds the local watchlist: whose new tweets
	// I want notifications about.
	SubscriptionsRepoName = "/SUBSCRIPTIONS"
)
View Source 
const TimelineRepoName = "/TIMELINE"
View Source 
const (
	TweetsNamespace = "/TWEETS"
)
View Source 
const WalletRepoName = "/WALLETS"

Variables

View Source 
var (
	ErrNilAliasesRepo = local_store.DBError("aliases repo is nil")
	ErrTooManyAliases = local_store.DBError("too many aliases")
	ErrAliasNotFound  = local_store.DBError("alias not found")
	ErrAliasRevoked   = local_store.DBError("pairing revoked")
)
View Source 
var (
	ErrChatNotFound    = local_store.DBError("chat not found")
	ErrMessageNotFound = local_store.DBError("message not found")
)
View Source 
var (
	ErrMediaNotFound    = local_store.DBError("media not found")
	ErrMediaRepoNotInit = local_store.DBError("media repo is not initialized")
)
View Source 
var (
	ErrTweetNotFound = local.DBError("tweet not found")
	ErrViewsNotFound = local.DBError("views not found")
	ErrReplyNotFound = local.DBError("reply not found")
)
View Source 
var (
	ErrUserNotFound      = local_store.DBError("user not found")
	ErrUserAlreadyExists = local_store.DBError("user already exists")
	ErrConflict          = local_store.ErrConflict
)
View Source 
var ErrAlreadyFollowed = local_store.DBError("already followed")
View Source 
var ErrFilterNotFound = local_store.DBError("filter not found")
View Source 
var ErrNilAuthRepo = local_store.DBError("auth repo is nil")
View Source 
var ErrNotificationsNotFound = local_store.DBError("notifications not found")
View Source 
var ErrOrderNotFound = local_store.DBError("order not found")
View Source 
var ErrPollAlreadyVoted = local_store.DBError("poll already voted")
View Source 
var ErrReactionsNotFound = local_store.DBError("reaction not found")
View Source 
var ErrWalletAddressNotFound = local_store.DBError("wallet address not found")

Functions

func NewRatingRepo added in v0.8.77 

func NewRatingRepo(db RatingStorer) ds.Datastore

NewRatingRepo backs the rating CRDT. It is kept apart from the stats CRDT because each go-ds-crdt instance owns its whole namespace.

func NewStatsRepo added in v0.6.120 

func NewStatsRepo(db StatsStorer) ds.Datastore

Types

type AliasesRepo added in v0.8.77 

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

func NewAliasesRepo added in v0.8.77 

func NewAliasesRepo(db AliasesStorer) *AliasesRepo

func (*AliasesRepo) DeleteAlias added in v0.8.77 

func (repo *AliasesRepo) DeleteAlias(nodeId string) error

DeleteAlias unpairs a device. Authorization ends with the record — the middleware reads the alias set on every private request — but the device keeps a usable pairing payload, and the session token that payload carries stays valid for as long as this node process runs. So the tombstone is kept without a TTL: it outlives an active alias's 72h window on purpose, and only a later SetAlias for the same node id with a different token (a fresh login's session token) ever overwrites it.

func (*AliasesRepo) GetAliases added in v0.8.77 

func (repo *AliasesRepo) GetAliases() (aliases []domain.Alias, err error)

func (*AliasesRepo) GetNodeIDs added in v0.8.77 

func (repo *AliasesRepo) GetNodeIDs() (ids []string, err error)

func (*AliasesRepo) SetAlias added in v0.8.77 

func (repo *AliasesRepo) SetAlias(alias domain.Alias) error

SetAlias pairs a device or refreshes the one already paired under the same node id. A device owns a single record, so a pair refresh renews its TTL instead of claiming another slot.

type AliasesStorer added in v0.8.77 

type AliasesStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type AuthRepo 

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

func NewAuthRepo 

func NewAuthRepo(db AuthStorer, network string) *AuthRepo

func (*AuthRepo) Authenticate 

func (repo *AuthRepo) Authenticate(username, password string) (err error)

func (*AuthRepo) GetOwner 

func (repo *AuthRepo) GetOwner() domain.Owner

func (*AuthRepo) Logout added in v0.4.37 

func (repo *AuthRepo) Logout()

func (*AuthRepo) PrivateKey 

func (repo *AuthRepo) PrivateKey() ed25519.PrivateKey

func (*AuthRepo) SessionToken 

func (repo *AuthRepo) SessionToken() string

func (*AuthRepo) SetOwner 

func (repo *AuthRepo) SetOwner(o domain.Owner) (_ domain.Owner, err error)

type AuthStorer 

type AuthStorer interface {
	Run(username, password string) (err error)
	Set(key local_store.DatabaseKey, value []byte) error
	Get(key local_store.DatabaseKey) ([]byte, error)
	NewTxn() (local_store.WarpTransactioner, error)
	Close()
}

type BlockLevel added in v0.5.105 

type BlockLevel int
const (
	InitialBlock BlockLevel = iota + 1
	MediumBlock
	AdvancedBlock
	PermanentBlock
)

func (BlockLevel) Next added in v0.5.105 

func (b BlockLevel) Next() BlockLevel

type BlocklistTerm added in v0.5.129 

type BlocklistTerm struct {
	PeerID string
	Level  BlockLevel
}

type BlocksRepo added in v0.7.26 

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

BlocksRepo persists the set of user ids the local owner has blocked.

func NewBlocksRepo added in v0.7.26 

func NewBlocksRepo(db BlocksStorer) *BlocksRepo

func (*BlocksRepo) Block added in v0.7.26 

func (repo *BlocksRepo) Block(blockerId, blockeeId string) error

Block records that blockerId has blocked blockeeId.

func (*BlocksRepo) IsBlocked added in v0.7.26 

func (repo *BlocksRepo) IsBlocked(blockerId, blockeeId string) (bool, error)

func (*BlocksRepo) List added in v0.7.26 

func (repo *BlocksRepo) List(blockerId string, limit *uint64, cursor *string) ([]string, string, error)

func (*BlocksRepo) Unblock added in v0.7.26 

func (repo *BlocksRepo) Unblock(blockerId, blockeeId string) error

Unblock removes a previously-recorded block.

type BlocksStorer added in v0.7.26 

type BlocksStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type BookmarkRepo added in v0.7.26 

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

func NewBookmarkRepo added in v0.7.26 

func NewBookmarkRepo(db BookmarkStorer) *BookmarkRepo

func (*BookmarkRepo) Bookmark added in v0.7.26 

func (repo *BookmarkRepo) Bookmark(userId, tweetId, ownerUserId string) error

func (*BookmarkRepo) List added in v0.7.26 

func (repo *BookmarkRepo) List(userId string, limit *uint64, cursor *string) ([]domain.Bookmark, string, error)

func (*BookmarkRepo) Unbookmark added in v0.7.26 

func (repo *BookmarkRepo) Unbookmark(userId, tweetId string) error

type BookmarkStorer added in v0.7.26 

type BookmarkStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type ChatRepo 

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

func NewChatRepo 

func NewChatRepo(db ChatStorer) *ChatRepo

func (*ChatRepo) CreateChat 

func (repo *ChatRepo) CreateChat(chatId *string, ownerId, otherUserId string) (chat domain.Chat, err error)

func (*ChatRepo) CreateMessage 

func (repo *ChatRepo) CreateMessage(msg domain.ChatMessage) (domain.ChatMessage, error)

func (*ChatRepo) DeleteChat 

func (repo *ChatRepo) DeleteChat(chatId string) error

func (*ChatRepo) DeleteMessage 

func (repo *ChatRepo) DeleteMessage(chatId, id string) error

func (*ChatRepo) GetChat 

func (repo *ChatRepo) GetChat(chatId string) (chat domain.Chat, err error)

func (*ChatRepo) GetMessage 

func (repo *ChatRepo) GetMessage(chatId, id string) (m domain.ChatMessage, err error)

func (*ChatRepo) GetUserChats 

func (repo *ChatRepo) GetUserChats(userId string, limit *uint64, cursor *string) ([]domain.Chat, string, error)

func (*ChatRepo) IsParticipants added in v0.8.77 

func (repo *ChatRepo) IsParticipants(ownerId, otherUserId string) bool

func (*ChatRepo) ListMessages 

func (repo *ChatRepo) ListMessages(chatId string, limit *uint64, cursor *string) ([]domain.ChatMessage, string, error)

type ChatStorer 

type ChatStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type FilterRepo added in v0.7.26 

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

func NewFilterRepo added in v0.7.26 

func NewFilterRepo(db FilterStorer) *FilterRepo

func (*FilterRepo) AddKeyword added in v0.7.26 

func (repo *FilterRepo) AddKeyword(userId, filterId string, kw domain.FilterKeyword) (domain.FilterKeyword, error)

AddKeyword appends a new keyword to an existing filter.

func (*FilterRepo) Create added in v0.7.26 

func (repo *FilterRepo) Create(userId string, f domain.Filter) (domain.Filter, error)

func (*FilterRepo) Delete added in v0.7.26 

func (repo *FilterRepo) Delete(userId, filterId string) error

func (*FilterRepo) DeleteKeyword added in v0.7.26 

func (repo *FilterRepo) DeleteKeyword(userId, keywordId string) error

DeleteKeyword removes a keyword by id from whichever filter owns it.

func (*FilterRepo) Get added in v0.7.26 

func (repo *FilterRepo) Get(userId, filterId string) (domain.Filter, error)

func (*FilterRepo) List added in v0.7.26 

func (repo *FilterRepo) List(userId string, limit *uint64, cursor *string) ([]domain.Filter, string, error)

func (*FilterRepo) Update added in v0.7.26 

func (repo *FilterRepo) Update(userId string, f domain.Filter) (domain.Filter, error)

func (*FilterRepo) UpdateKeyword added in v0.7.26 

func (repo *FilterRepo) UpdateKeyword(userId string, kw domain.FilterKeyword) (domain.FilterKeyword, error)

UpdateKeyword replaces a keyword on the filter that owns it.

type FilterStorer added in v0.7.26 

type FilterStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type FollowRepo 

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

func NewFollowRepo 

func NewFollowRepo(db FollowerStorer) *FollowRepo

func (*FollowRepo) AddFollowRequest added in v0.7.26 

func (repo *FollowRepo) AddFollowRequest(targetUserId, followerId string) error

AddFollowRequest records a pending follow request from followerId against targetUserId (locked-user flow). On approval the entry should be removed via RemoveFollowRequest and re-recorded via Follow; on reject it's just removed.

func (*FollowRepo) Follow 

func (repo *FollowRepo) Follow(fromUserId, toUserId string) error

func (*FollowRepo) GetFollowers 

func (repo *FollowRepo) GetFollowers(userId string, limit *uint64, cursor *string) ([]string, string, error)

func (*FollowRepo) GetFollowersCount 

func (repo *FollowRepo) GetFollowersCount(userId string) (uint64, error)

func (*FollowRepo) GetFollowings added in v0.5.83 

func (repo *FollowRepo) GetFollowings(userId string, limit *uint64, cursor *string) ([]string, string, error)

GetFollowings following - one who is followed (has his/her posts monitored by another user)

func (*FollowRepo) GetFollowingsCount added in v0.5.83 

func (repo *FollowRepo) GetFollowingsCount(userId string) (uint64, error)

func (*FollowRepo) HasFollowRequest added in v0.7.26 

func (repo *FollowRepo) HasFollowRequest(targetUserId, followerId string) (bool, error)

HasFollowRequest reports whether followerId has an outstanding pending follow request against targetUserId.

func (*FollowRepo) IsFollower added in v0.5.83 

func (repo *FollowRepo) IsFollower(ownerId, otherUserId string) bool

func (*FollowRepo) IsFollowing added in v0.5.83 

func (repo *FollowRepo) IsFollowing(ownerId, otherUserId string) bool

func (*FollowRepo) ListFollowRequests added in v0.7.26 

func (repo *FollowRepo) ListFollowRequests(targetUserId string, limit *uint64, cursor *string) ([]string, string, error)

ListFollowRequests returns the follower ids waiting for approval on the given target user, oldest-first.

func (*FollowRepo) RemoveFollowRequest added in v0.7.26 

func (repo *FollowRepo) RemoveFollowRequest(targetUserId, followerId string) error

RemoveFollowRequest deletes a pending follow request (idempotent).

func (*FollowRepo) Unfollow 

func (repo *FollowRepo) Unfollow(fromUserId, toUserId string) error

type FollowerStorer 

type FollowerStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
	Set(key local_store.DatabaseKey, value []byte) error
	Get(key local_store.DatabaseKey) ([]byte, error)
	Delete(key local_store.DatabaseKey) error
}

type MediaRepo 

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

func NewChatMediaRepo added in v0.8.77 

func NewChatMediaRepo(db MediaStorer) *MediaRepo

func NewMediaRepo 

func NewMediaRepo(db MediaStorer) *MediaRepo

func NewSponsoredMediaRepo added in v0.8.77 

func NewSponsoredMediaRepo(db MediaStorer) *MediaRepo

func (*MediaRepo) GetCopy added in v0.8.77 

func (repo *MediaRepo) GetCopy(userId, key string) (domain.MediaCopy, error)

func (*MediaRepo) GetImage 

func (repo *MediaRepo) GetImage(userId, key string) (domain.Base64Image, error)

func (*MediaRepo) GetVideo added in v0.8.77 

func (repo *MediaRepo) GetVideo(userId, key string) (domain.Base64Video, error)

func (*MediaRepo) ListCopies added in v0.8.77 

func (repo *MediaRepo) ListCopies(userId, originalKey string) ([]domain.MediaCopy, error)

func (*MediaRepo) SetCopy added in v0.8.77 

func (repo *MediaRepo) SetCopy(userId, key string, c domain.MediaCopy) error

func (*MediaRepo) SetForeignImageWithTTL 

func (repo *MediaRepo) SetForeignImageWithTTL(userId, key string, img domain.Base64Image) error

func (*MediaRepo) SetForeignVideoWithTTL added in v0.8.77 

func (repo *MediaRepo) SetForeignVideoWithTTL(userId, key string, video domain.Base64Video) error

func (*MediaRepo) SetImage 

func (repo *MediaRepo) SetImage(userId string, img domain.Base64Image) (_ domain.ImageKey, err error)

func (*MediaRepo) SetVideo added in v0.8.77 

func (repo *MediaRepo) SetVideo(userId string, video domain.Base64Video) (_ domain.VideoKey, err error)

type MediaStorer 

type MediaStorer interface {
	Set(key local_store.DatabaseKey, value []byte) error
	Get(key local_store.DatabaseKey) ([]byte, error)
	SetWithTTL(key local_store.DatabaseKey, value []byte, ttl time.Duration) error
	NewReadTxn() (local_store.WarpTransactioner, error)
}

type MutesRepo added in v0.7.26 

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

MutesRepo persists the set of user ids the local owner has muted.

func NewMutesRepo added in v0.7.26 

func NewMutesRepo(db MutesStorer) *MutesRepo

func (*MutesRepo) IsMuted added in v0.7.26 

func (repo *MutesRepo) IsMuted(muterId, muteeId string) (bool, error)

func (*MutesRepo) List added in v0.7.26 

func (repo *MutesRepo) List(muterId string, limit *uint64, cursor *string) ([]string, string, error)

func (*MutesRepo) Mute added in v0.7.26 

func (repo *MutesRepo) Mute(muterId, muteeId string) error

func (*MutesRepo) Unmute added in v0.7.26 

func (repo *MutesRepo) Unmute(muterId, muteeId string) error

type MutesStorer added in v0.7.26 

type MutesStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type NewUserNotifier added in v0.7.483 

type NewUserNotifier interface {
	Add(not domain.Notification) error
}

NewUserNotifier records a "new user discovered" notification for the local owner when a previously-unknown user is first stored.

type NodeRepo 

type NodeRepo struct {
	RelaySelfHashHex string
	// contains filtered or unexported fields
}

func NewNodeRepo 

func NewNodeRepo(db NodeStorer) *NodeRepo

func (*NodeRepo) Batch 

func (d *NodeRepo) Batch(ctx context.Context) (datastore.Batch, error)

func (*NodeRepo) BlocklistExponential added in v0.3.9 

func (d *NodeRepo) BlocklistExponential(peerId string) error

func (*NodeRepo) BlocklistPermanent added in v0.7.26 

func (d *NodeRepo) BlocklistPermanent(peerId string) error

BlocklistPermanent puts peerId on the peer-level blocklist with no TTL (PermanentBlock). Unlike Blocklist (which escalates the BlockLevel and writes the matching expiring entry), this is what social blocks resolve to: the user explicitly decided to block, and the ban stays until they unblock.

func (*NodeRepo) BlocklistRemove 

func (d *NodeRepo) BlocklistRemove(peerId string) error

func (*NodeRepo) BlocklistTerm added in v0.5.129 

func (d *NodeRepo) BlocklistTerm(peerId string) (*BlocklistTerm, error)

func (*NodeRepo) Close 

func (d *NodeRepo) Close() (err error)

func (*NodeRepo) Delete 

func (d *NodeRepo) Delete(ctx context.Context, key datastore.Key) error

func (*NodeRepo) DiskUsage 

func (d *NodeRepo) DiskUsage(ctx context.Context) (uint64, error)

DiskUsage implements the PersistentDatastore interface. It returns the sum of lsm and value log files sizes in bytes.

func (*NodeRepo) Get 

func (d *NodeRepo) Get(ctx context.Context, key datastore.Key) (value []byte, err error)

func (*NodeRepo) GetExpiration 

func (d *NodeRepo) GetExpiration(ctx context.Context, key datastore.Key) (t time.Time, err error)

func (*NodeRepo) GetSize 

func (d *NodeRepo) GetSize(ctx context.Context, key datastore.Key) (_ int, err error)

func (*NodeRepo) Has 

func (d *NodeRepo) Has(ctx context.Context, key datastore.Key) (_ bool, err error)

func (*NodeRepo) IsBlocklisted 

func (d *NodeRepo) IsBlocklisted(peerId string) bool

func (*NodeRepo) Put 

func (d *NodeRepo) Put(ctx context.Context, key datastore.Key, value []byte) error

func (*NodeRepo) PutWithTTL 

func (d *NodeRepo) PutWithTTL(ctx context.Context, key datastore.Key, value []byte, ttl time.Duration) error

func (*NodeRepo) Query 

func (d *NodeRepo) Query(ctx context.Context, q datastore.Query) (datastore.Results, error)

func (*NodeRepo) SetTTL 

func (d *NodeRepo) SetTTL(ctx context.Context, key datastore.Key, ttl time.Duration) error

func (*NodeRepo) Sync 

func (d *NodeRepo) Sync(ctx context.Context, _ datastore.Key) error

type NodeStorer 

type NodeStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
	Get(key local_store.DatabaseKey) ([]byte, error)
	GetExpiration(key local_store.DatabaseKey) (uint64, error)
	GetSize(key local_store.DatabaseKey) (int64, error)
	Sync() error
	IsClosed() bool
	InnerDB() *local_store.WarpDB
	SetWithTTL(key local_store.DatabaseKey, value []byte, ttl time.Duration) error
	Set(key local_store.DatabaseKey, value []byte) error
	Delete(key local_store.DatabaseKey) error
}

type NotificationsRepo added in v0.5.30 

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

func NewNotificationsRepo added in v0.5.30 

func NewNotificationsRepo(db NotificationsStorer) *NotificationsRepo

func (*NotificationsRepo) Add added in v0.5.30 

func (repo *NotificationsRepo) Add(not domain.Notification) error

func (*NotificationsRepo) Get added in v0.7.26 

func (repo *NotificationsRepo) Get(userId, notificationId string) (domain.Notification, error)

func (*NotificationsRepo) List added in v0.5.30 

func (repo *NotificationsRepo) List(userId string, limit *uint64, cursor *string) ([]domain.Notification, string, error)

func (*NotificationsRepo) MarkAllRead added in v0.7.422 

func (repo *NotificationsRepo) MarkAllRead(userId string) error

MarkAllRead flips IsRead to true for every unread notification the user has. Unread keys are collected in a read-only txn (no conflict tracking), then each is flipped with the same targeted-write pattern MarkRead uses — one small txn per key — so concurrent single MarkRead calls can't commit-conflict a whole page. IsRead is monotonic, so racing writers converge on the same state. Rows expired by TTL between the scan and the write are skipped.

func (*NotificationsRepo) MarkRead added in v0.7.26 

func (repo *NotificationsRepo) MarkRead(userId, notificationId string) error

MarkRead flips Notification.IsRead to true for the given notification. Notification keys are timestamp-indexed within a per-user prefix, so a scan is unavoidable to locate the row from (userId, notificationId).

The scan and the write live in *separate* transactions on purpose. Badger's SSI tracks every key the txn reads; doing the prefix scan inside the same RW txn that later writes one key would add the entire page (~100 sibling notifications) to the read set, and a concurrent MarkRead on any of those siblings would commit-conflict the second writer. Splitting the work means the write txn's read+write sets are both just {targetKey}, so two MarkRead calls on different notifications no longer collide. Two MarkRead calls on the *same* notification can still race, but IsRead is monotonic, so the loser's view-after-commit matches the winner's regardless.

func (*NotificationsRepo) ReverseList added in v0.7.362 

func (repo *NotificationsRepo) ReverseList(userId string, cursor *string, limit *uint64) ([]domain.Notification, string, error)

func (*NotificationsRepo) UnreadCount added in v0.7.59 

func (repo *NotificationsRepo) UnreadCount(userId string) (uint64, error)

UnreadCount scans every notification under the user's prefix and returns the number with IsRead == false. List paginates, so the caller can't count "unread across all pages" without doing the full scan itself; this method centralises that walk so handlers don't derive the unread count from one page (which gave a flickering "20 unread" badge that mirrored whatever happened to be on page 1).

O(N) over the user's stored notifications. The repo's 24 h TTL on every Add() bounds N, so a scan per call is acceptable for now; if volume grows we'll move to a maintained counter mirrored on Add / MarkRead.

type NotificationsStorer added in v0.5.30 

type NotificationsStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
	NewReadTxn() (local_store.WarpTransactioner, error)
}

type OrderRepo added in v0.8.77 

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

func NewOrderRepo added in v0.8.77 

func NewOrderRepo(db OrderStorer) *OrderRepo

func (*OrderRepo) CountConfirmed added in v0.8.77 

func (repo *OrderRepo) CountConfirmed(buyerId string, from, to time.Time) (int, error)

func (*OrderRepo) Delete added in v0.8.77 

func (repo *OrderRepo) Delete(tweetId, buyerId string) error

func (*OrderRepo) Get added in v0.8.77 

func (repo *OrderRepo) Get(tweetId, buyerId string) (domain.Order, error)

func (*OrderRepo) Save added in v0.8.77 

func (repo *OrderRepo) Save(o domain.Order) error

type OrderStorer added in v0.8.77 

type OrderStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type OutboxRepo added in v0.7.521 

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

func NewOutboxRepo added in v0.7.521 

func NewOutboxRepo(db OutboxStorer) *OutboxRepo

func (*OutboxRepo) Delete added in v0.7.521 

func (repo *OutboxRepo) Delete(destNodeId, messageId string) error

func (*OutboxRepo) Enqueue added in v0.7.521 

func (repo *OutboxRepo) Enqueue(destNodeId, route string, payload []byte) (event.Message, error)

func (*OutboxRepo) ListByNode added in v0.7.521 

func (repo *OutboxRepo) ListByNode(destNodeId string) ([]event.Message, error)

func (*OutboxRepo) ListNodes added in v0.7.521 

func (repo *OutboxRepo) ListNodes() ([]string, error)

type OutboxStorer added in v0.7.521 

type OutboxStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type PollRepo added in v0.8.77 

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

func NewPollRepo added in v0.8.77 

func NewPollRepo(db PollStorer, statsDb PollStatsStorer) *PollRepo

func (*PollRepo) Results added in v0.8.77 

func (repo *PollRepo) Results(tweetId string, optionsNum int) (votes []uint64, err error)

func (*PollRepo) Vote added in v0.8.77 

func (repo *PollRepo) Vote(tweetId, userId string, option int, isTransitive bool) error

Vote records userId's choice on the poll attached to tweetId. A vote is final: a second call for the same voter returns ErrPollAlreadyVoted and changes nothing, so a replayed or propagated event can't inflate a count.

isTransitive carries the same meaning as in ReactionRepo: the network-wide (CRDT) counter is bumped only on the voter's own node, so a vote stored on both the voter's and the author's node is counted once.

func (*PollRepo) Voted added in v0.8.77 

func (repo *PollRepo) Voted(tweetId, userId string) (option int, ok bool, err error)

Voted returns the option userId picked on tweetId's poll. ok is false when they haven't voted.

type PollStatsStorer added in v0.8.77 

type PollStatsStorer interface {
	GetAggregatedStat(key ds.Key) (uint64, error)
	Increment(key ds.Key) error
}

type PollStorer added in v0.8.77 

type PollStorer interface {
	Get(key local_store.DatabaseKey) ([]byte, error)
	NewTxn() (local_store.WarpTransactioner, error)
}

type RatingStorer added in v0.8.77 

type RatingStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
	Get(key local_store.DatabaseKey) ([]byte, error)
	GetExpiration(key local_store.DatabaseKey) (uint64, error)
	GetSize(key local_store.DatabaseKey) (int64, error)
	Sync() error
	IsClosed() bool
	InnerDB() *local_store.WarpDB
	SetWithTTL(key local_store.DatabaseKey, value []byte, ttl time.Duration) error
	Set(key local_store.DatabaseKey, value []byte) error
	Delete(key local_store.DatabaseKey) error
}

type ReactionRepo added in v0.8.77 

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

func NewReactionRepo added in v0.8.77 

func NewReactionRepo(db ReactionStorer, statsDb ReactionStatsStorer) *ReactionRepo

func (*ReactionRepo) React added in v0.8.77 

func (repo *ReactionRepo) React(tweetId, userId, emoji string, isTransitive bool) (reactionsCount uint64, err error)

func (*ReactionRepo) Reacted added in v0.8.77 

func (repo *ReactionRepo) Reacted(userId string, limit *uint64, cursor *string) ([]domain.ReactedTweet, string, error)

func (*ReactionRepo) Reaction added in v0.8.77 

func (repo *ReactionRepo) Reaction(tweetId, userId string) (string, error)

func (*ReactionRepo) Reactions added in v0.8.77 

func (repo *ReactionRepo) Reactions(tweetId string) (map[string]uint64, error)

func (*ReactionRepo) ReactionsCount added in v0.8.77 

func (repo *ReactionRepo) ReactionsCount(tweetId string) (reactionsNum uint64, err error)

func (*ReactionRepo) Reactors added in v0.8.77 

func (repo *ReactionRepo) Reactors(tweetId string, limit *uint64, cursor *string) (_ reactorIDs, cur string, err error)

func (*ReactionRepo) RemoveReacted added in v0.8.77 

func (repo *ReactionRepo) RemoveReacted(userId, tweetId string) error

func (*ReactionRepo) SetReacted added in v0.8.77 

func (repo *ReactionRepo) SetReacted(userId, tweetId, ownerUserId string) error

func (*ReactionRepo) Unreact added in v0.8.77 

func (repo *ReactionRepo) Unreact(tweetId, userId string, isTransitive bool) (reactionsCount uint64, err error)

type ReactionStatsStorer added in v0.8.77 

type ReactionStatsStorer interface {
	GetAggregatedStat(key ds.Key) (uint64, error)
	Increment(key ds.Key) error
	Decrement(key ds.Key) error
}

type ReactionStorer added in v0.8.77 

type ReactionStorer interface {
	Get(key local_store.DatabaseKey) ([]byte, error)
	NewTxn() (local_store.WarpTransactioner, error)
}

type SettingsRepo added in v0.7.483 

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

func NewSettingsRepo added in v0.7.483 

func NewSettingsRepo(db SettingsStorer) *SettingsRepo

func (*SettingsRepo) GetGatewaySettings added in v0.8.77 

func (repo *SettingsRepo) GetGatewaySettings(userId string) (domain.GatewaySettings, error)

GetGatewaySettings returns the user's gateway settings, or a zero-value (empty NodeID) record when none has been saved yet.

func (*SettingsRepo) GetNotificationSettings added in v0.7.483 

func (repo *SettingsRepo) GetNotificationSettings(userId string) (domain.NotificationSettings, error)

GetNotificationSettings returns the user's notification settings, or a zero-value (email disabled) record when none has been saved yet.

func (*SettingsRepo) GetRateLimitSettings added in v0.8.77 

func (repo *SettingsRepo) GetRateLimitSettings(userId string) (ratelimit.Settings, error)

func (*SettingsRepo) SetGatewaySettings added in v0.8.77 

func (repo *SettingsRepo) SetGatewaySettings(userId string, s domain.GatewaySettings) error

SetGatewaySettings persists the user's gateway settings.

func (*SettingsRepo) SetNotificationSettings added in v0.7.483 

func (repo *SettingsRepo) SetNotificationSettings(userId string, s domain.NotificationSettings) error

SetNotificationSettings persists the user's notification settings.

func (*SettingsRepo) SetRateLimitSettings added in v0.8.77 

func (repo *SettingsRepo) SetRateLimitSettings(userId string, s ratelimit.Settings) error

type SettingsStorer added in v0.7.483 

type SettingsStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type StatsStorer added in v0.6.120 

type StatsStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
	Get(key local_store.DatabaseKey) ([]byte, error)
	GetExpiration(key local_store.DatabaseKey) (uint64, error)
	GetSize(key local_store.DatabaseKey) (int64, error)
	Sync() error
	IsClosed() bool
	InnerDB() *local_store.WarpDB
	SetWithTTL(key local_store.DatabaseKey, value []byte, ttl time.Duration) error
	Set(key local_store.DatabaseKey, value []byte) error
	Delete(key local_store.DatabaseKey) error
}

type SubscriptionsRepo added in v0.7.26 

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

SubscriptionsRepo persists the local "notify me about this user's new posts" watchlist.

func NewSubscriptionsRepo added in v0.7.26 

func NewSubscriptionsRepo(db SubscriptionsStorer) *SubscriptionsRepo

func (*SubscriptionsRepo) IsSubscribed added in v0.7.26 

func (repo *SubscriptionsRepo) IsSubscribed(selfId, targetUserId string) (bool, error)

func (*SubscriptionsRepo) Subscribe added in v0.7.26 

func (repo *SubscriptionsRepo) Subscribe(selfId, targetUserId string) error

func (*SubscriptionsRepo) Unsubscribe added in v0.7.26 

func (repo *SubscriptionsRepo) Unsubscribe(selfId, targetUserId string) error

type SubscriptionsStorer added in v0.7.26 

type SubscriptionsStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

type TimelineRepo 

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

func NewTimelineRepo 

func NewTimelineRepo(db TimelineStorer) *TimelineRepo

func (*TimelineRepo) AddTweetToTimeline 

func (repo *TimelineRepo) AddTweetToTimeline(userId string, tweet domain.Tweet) error

func (*TimelineRepo) DeleteTweetFromTimeline 

func (repo *TimelineRepo) DeleteTweetFromTimeline(userID, tweetID string) error

func (*TimelineRepo) GetTimeline 

func (repo *TimelineRepo) GetTimeline(userId string, limit *uint64, cursor *string) ([]domain.Tweet, string, error)

GetTimeline retrieves a user's timeline sorted from newest to oldest

type TimelineStorer 

type TimelineStorer interface {
	Set(key local_store.DatabaseKey, value []byte) error
	NewTxn() (local_store.WarpTransactioner, error)
	Delete(key local_store.DatabaseKey) error
}

type TweetRepo 

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

func NewTweetRepo 

func NewTweetRepo(db TweetsStorer, statsDb TweetStatsStorer) *TweetRepo

func (*TweetRepo) AddReply added in v0.7.381 

func (repo *TweetRepo) AddReply(reply domain.Tweet, isTransitive bool) (domain.Tweet, error)

isTransitive tells whether this reply should propagate to the network-wide (CRDT) reply counter, which is replicated ("transits") across nodes. The handler sets it true only on the replier's own node, so a reply stored on both the replier's and the parent author's node is counted once. The local per-node counter is always updated (it backs the read-time fallback).

func (*TweetRepo) AppendEdit added in v0.7.26 

func (repo *TweetRepo) AppendEdit(edit domain.TweetEdit) (domain.TweetEdit, error)

func (*TweetRepo) Blocklist added in v0.5.30 

func (repo *TweetRepo) Blocklist(tweetId string) error

func (*TweetRepo) Create 

func (repo *TweetRepo) Create(userId string, tweet domain.Tweet) (domain.Tweet, error)

Create adds a new tweet to the database

func (*TweetRepo) CreateWithTTL added in v0.3.61 

func (repo *TweetRepo) CreateWithTTL(userId string, tweet domain.Tweet, duration time.Duration) (domain.Tweet, error)

func (*TweetRepo) Delete 

func (repo *TweetRepo) Delete(userID, tweetID string) error

Delete removes a tweet by its ID

func (*TweetRepo) DeleteReply added in v0.7.381 

func (repo *TweetRepo) DeleteReply(parentID, replyID string, isTransitive bool) (domain.Tweet, error)

func (*TweetRepo) Get 

func (repo *TweetRepo) Get(userID, tweetID string) (tweet domain.Tweet, err error)

Get retrieves a tweet by its ID

func (*TweetRepo) GetReplies added in v0.7.381 

func (repo *TweetRepo) GetReplies(parentID string, limit *uint64, cursor *string) ([]domain.Tweet, string, error)

GetReplies returns the direct replies to parentID, newest first — the tweet List of the parent's partition. Replies are tweets, so the result is a flat []domain.Tweet, each carrying its ParentId for clients that want to nest.

func (*TweetRepo) GetReply added in v0.7.381 

func (repo *TweetRepo) GetReply(parentID, replyID string) (domain.Tweet, error)

func (*TweetRepo) GetViewsCount added in v0.5.158 

func (repo *TweetRepo) GetViewsCount(tweetId string) (uint64, error)

func (*TweetRepo) IsBlocklisted added in v0.5.30 

func (repo *TweetRepo) IsBlocklisted(tweetId string) bool

func (*TweetRepo) List 

func (repo *TweetRepo) List(userId string, limit *uint64, cursor *string) ([]domain.Tweet, string, error)

func (*TweetRepo) NewRetweet 

func (repo *TweetRepo) NewRetweet(tweet domain.Tweet, isTransitive bool) (_ domain.Tweet, err error)

func (*TweetRepo) Pin added in v0.7.26 

func (repo *TweetRepo) Pin(userId, tweetId string) (domain.Tweet, error)

func (*TweetRepo) RecordView added in v0.6.316 

func (repo *TweetRepo) RecordView(tweetId, viewerId string) (uint64, error)

RecordView increments the view counter for tweetId on behalf of viewerId. The (tweetId, viewerId) pair is recorded permanently, so subsequent views from the same viewer — across sessions, restarts, days — are no-ops. The first call wins; the counter is incremented exactly once per unique viewer. The increment is atomic via the underlying transaction and replicated through the CRDT stats store, so it is safe under concurrent calls across nodes.

func (*TweetRepo) RepliesCount added in v0.7.381 

func (repo *TweetRepo) RepliesCount(tweetId string) (uint64, error)

func (*TweetRepo) Retweeters 

func (repo *TweetRepo) Retweeters(tweetId string, limit *uint64, cursor *string) (_ retweetersIDs, cur string, err error)

func (*TweetRepo) RetweetsCount 

func (repo *TweetRepo) RetweetsCount(tweetId string) (uint64, error)

func (*TweetRepo) TweetsCount 

func (repo *TweetRepo) TweetsCount(userId string) (uint64, error)

func (*TweetRepo) UnRetweet 

func (repo *TweetRepo) UnRetweet(retweetedByUserID, tweetId string, isTransitive bool) error

func (*TweetRepo) Unpin added in v0.7.26 

func (repo *TweetRepo) Unpin(userId, tweetId string) (domain.Tweet, error)

func (*TweetRepo) Update added in v0.5.134 

func (repo *TweetRepo) Update(updateTweet domain.Tweet) error

type TweetStatsStorer added in v0.5.158 

type TweetStatsStorer interface {
	GetAggregatedStat(key ds.Key) (uint64, error)
	Increment(key ds.Key) error
	Decrement(key ds.Key) error
}

type TweetsStorer 

type TweetsStorer interface {
	NewTxn() (local.WarpTransactioner, error)
	Set(key local.DatabaseKey, value []byte) error
	Get(key local.DatabaseKey) ([]byte, error)
	Delete(key local.DatabaseKey) error
	GetExpiration(key local.DatabaseKey) (uint64, error)
}

type UserRepo 

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

func NewUserRepo 

func NewUserRepo(db UserStorer) *UserRepo

func NewUserRepoNotifying added in v0.7.483 

func NewUserRepoNotifying(db UserStorer, notifier NewUserNotifier, ownerUserId string) *UserRepo

NewUserRepoNotifying returns a UserRepo that, on first storing a genuinely new (non-owner) user, records a "new user discovered" notification for the owner via notifier.

func (*UserRepo) Create 

func (repo *UserRepo) Create(user domain.User) (domain.User, error)

Create adds a new user to the database

func (*UserRepo) CreateWithTTL added in v0.3.61 

func (repo *UserRepo) CreateWithTTL(user domain.User, ttl time.Duration) (domain.User, error)

func (*UserRepo) Delete 

func (repo *UserRepo) Delete(userId string) error

Delete removes a user by their ID

func (*UserRepo) Get 

func (repo *UserRepo) Get(userId string) (user domain.User, err error)

Get retrieves a user by their ID

func (*UserRepo) GetBatch 

func (repo *UserRepo) GetBatch(userIDs ...string) (users []domain.User, err error)

func (*UserRepo) GetByNodeID 

func (repo *UserRepo) GetByNodeID(nodeID string) (user domain.User, err error)

func (*UserRepo) List 

func (repo *UserRepo) List(limit *uint64, cursor *string) ([]domain.User, string, error)

func (*UserRepo) Search added in v0.7.26 

func (repo *UserRepo) Search(query string, limit *uint64, cursor *string) ([]domain.User, string, error)

Search returns users whose Username, Bio, or NodeId contains the (lower-cased) query. This is a server-side scan-and-filter — it preserves the cursor for incremental pages but still touches every matching record in the prefix range. A true substring index belongs here later; the API shape is forward-compatible.

func (*UserRepo) Update 

func (repo *UserRepo) Update(userId string, newUser domain.User) (domain.User, error)

func (*UserRepo) WhoToFollow added in v0.3.75 

func (repo *UserRepo) WhoToFollow(limit *uint64, cursor *string) ([]domain.User, string, error)

WhoToFollow returns an even share of every network the node knows users on. It is one page, so the returned cursor is always the end.

type UserStorer 

type UserStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
	Set(key local_store.DatabaseKey, value []byte) error
	Get(key local_store.DatabaseKey) ([]byte, error)
	Delete(key local_store.DatabaseKey) error
}

type WalletRepo added in v0.8.77 

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

func NewWalletRepo added in v0.8.77 

func NewWalletRepo(db WalletStorer) *WalletRepo

func (*WalletRepo) GetAddress added in v0.8.77 

func (repo *WalletRepo) GetAddress(chain, userId string) (domain.WalletAddress, error)

func (*WalletRepo) ListAddresses added in v0.8.77 

func (repo *WalletRepo) ListAddresses(chain string, limit *uint64, cursor *string) ([]domain.WalletAddress, string, error)

func (*WalletRepo) SetAddress added in v0.8.77 

func (repo *WalletRepo) SetAddress(chain, userId, address string) error

type WalletStorer added in v0.8.77 

type WalletStorer interface {
	NewTxn() (local_store.WarpTransactioner, error)
}

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.