database

package
v0.7.522 Latest Latest
Warning

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

 Go to latest Published: Jul 21, 2026 License: AGPL-3.0, AGPL-3.0-or-later Imports: 23 Imported by: 0

Documentation

Index

Constants

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 (
	LikeRepoName      = "/LIKES"
	IncrSubNamespace  = "INCR"
	LikerSubNamespace = "LIKER"
	LikedSubNamespace = "LIKED" // per-user index of liked tweet refs
)
View Source 
const (
	MediaRepoName     = "/MEDIA"
	ImageSubNamespace = "IMAGES"
	ImageMetaSubNS    = "IMAGES_META"
	// VideoSubNamespace is reserved — videos are not yet supported.
	VideoSubNamespace = "VIDEOS"
)
View Source 
const (
	BlocklistSubNamespace     = "BLOCKLIST"
	BlocklistUserSubNamespace = "USER"
	BlocklistTermSubNamespace = "TERM"
)
View Source 
const (
	UsersRepoName = "/USERS"

	DefaultWarpnetUserNetwork = "warpnet"
)
View Source 
const (
	BlocksRepoName = "/BLOCKS"
)
View Source 
const BookmarkRepoName = "/BOOKMARKS"
View Source 
const (
	DevicesRepoName = "/DEVICES"
)
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 OutboxNamespace = "/OUTBOX"

OutboxNamespace holds outgoing messages that could not be delivered because the destination node was offline. Entries are keyed per destination node and replayed once that node is seen online again.

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"
)

Variables

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")
)
View Source 
var ErrAlreadyFollowed = local_store.DBError("already followed")
View Source 
var ErrFilterNotFound = local_store.DBError("filter not found")
View Source 
var ErrLikesNotFound = local_store.DBError("like not found")
View Source 
var ErrNilAuthRepo = local_store.DBError("auth repo is nil")
View Source 
var ErrNilDevicesRepo = local_store.DBError("devices repo is nil")
View Source 
var ErrNotificationsNotFound = local_store.DBError("notifications not found")

Functions

func NewStatsRepo added in v0.6.120 

func NewStatsRepo(db StatsStorer) ds.Datastore

Types

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 Base64Image 

type Base64Image string

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) 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 DevicesRepo added in v0.6.263 

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

func NewDevicesRepo added in v0.6.263 

func NewDevicesRepo(db DevicesStorer) *DevicesRepo

func (*DevicesRepo) GetDevices added in v0.6.263 

func (repo *DevicesRepo) GetDevices(ownerNodeId string) (devices []domain.Device, err error)

func (*DevicesRepo) SetDevice added in v0.6.263 

func (repo *DevicesRepo) SetDevice(ownerNodeId string, device domain.Device) error

type DevicesStorer added in v0.6.263 

type DevicesStorer interface {
	SetWithTTL(key local_store.DatabaseKey, value []byte, ttl time.Duration) error
	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 ImageKey 

type ImageKey string

type LikeRepo 

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

func NewLikeRepo 

func NewLikeRepo(db LikeStorer, statsDb LikeStatsStorer) *LikeRepo

func (*LikeRepo) Like 

func (repo *LikeRepo) Like(tweetId, userId string, isTransitive bool) (likesCount uint64, err error)

isTransitive tells whether this action should propagate to the network-wide (CRDT) counter, which is replicated ("transits") across nodes. The caller (handler) sets it true only on the acting user's own node, so an action observed on more than one node is counted once. The local per-node counter is always updated (it backs the read-time fallback).

func (*LikeRepo) Liked added in v0.7.403 

func (repo *LikeRepo) Liked(userId string, limit *uint64, cursor *string) ([]domain.LikedTweet, string, error)

func (*LikeRepo) Likers 

func (repo *LikeRepo) Likers(tweetId string, limit *uint64, cursor *string) (_ likedUserIDs, cur string, err error)

func (*LikeRepo) LikesCount 

func (repo *LikeRepo) LikesCount(tweetId string) (likesNum uint64, err error)

func (*LikeRepo) RemoveLiked added in v0.7.403 

func (repo *LikeRepo) RemoveLiked(userId, tweetId string) error

func (*LikeRepo) SetLiked added in v0.7.403 

func (repo *LikeRepo) SetLiked(userId, tweetId, ownerUserId string) error

func (*LikeRepo) Unlike 

func (repo *LikeRepo) Unlike(tweetId, userId string, isTransitive bool) (likesCount uint64, err error)

type LikeStatsStorer added in v0.5.158 

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

type LikeStorer 

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

type MediaMeta added in v0.7.26 

type MediaMeta struct {
	Description string  `json:"description"`
	FocusX      float32 `json:"focus_x"`
	FocusY      float32 `json:"focus_y"`
}

MediaMeta is the per-image metadata layer: Mastodon's compose alt-text (description) and focal point (focus_x / focus_y in [-1, 1]). Stored under a parallel key from the image blob so updating metadata doesn't rewrite the (potentially MB-sized) base64 payload.

type MediaRepo 

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

func NewMediaRepo 

func NewMediaRepo(db MediaStorer) *MediaRepo

func (*MediaRepo) GetImage 

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

func (*MediaRepo) GetImageMeta added in v0.7.26 

func (repo *MediaRepo) GetImageMeta(userId, key string) (MediaMeta, error)

func (*MediaRepo) SetForeignImageWithTTL 

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

func (*MediaRepo) SetImage 

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

func (*MediaRepo) SetImageMeta added in v0.7.26 

func (repo *MediaRepo) SetImageMeta(userId, key string, meta MediaMeta) 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
}

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 OutboxEntry added in v0.7.521 

type OutboxEntry struct {
	Id         string    `json:"id"`
	DestNodeId string    `json:"dest_node_id"`
	Route      string    `json:"route"`
	Payload    []byte    `json:"payload"`
	Attempts   int       `json:"attempts"`
	CreatedAt  time.Time `json:"created_at"`
}

OutboxEntry is one undelivered outgoing stream request. Id is a ULID that doubles as the FIFO ordering key and as the stable envelope message id reused across redelivery attempts, so the receiver's idempotency layer can dedupe.

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, id string) error

Delete removes a delivered entry.

func (*OutboxRepo) Enqueue added in v0.7.521 

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

Enqueue stores an undelivered request for destNodeId and returns the created entry. The ULID key preserves per-node FIFO order for later replay.

func (*OutboxRepo) ListByNode added in v0.7.521 

func (repo *OutboxRepo) ListByNode(destNodeId string) ([]OutboxEntry, error)

ListByNode returns every queued entry for destNodeId in FIFO (oldest-first) order.

func (*OutboxRepo) ListNodes added in v0.7.521 

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

ListNodes returns the distinct destination node ids that currently have at least one queued entry, used to flush everything on startup.

func (*OutboxRepo) Save added in v0.7.521 

func (repo *OutboxRepo) Save(entry OutboxEntry) error

Save rewrites an existing entry in place (same key), used to persist an incremented attempt counter after a failed redelivery.

type OutboxStorer added in v0.7.521 

type OutboxStorer interface {
	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) 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) SetNotificationSettings added in v0.7.483 

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

SetNotificationSettings persists the user's notification settings.

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)

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
}

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.