repos

Documentation

Index

• type BackfillJobFinishItem
• type BackfillRepo
  • func NewBackfillRepo(db model.FederationDatabase, cache service.Cache) *BackfillRepo
  • func (r *BackfillRepo) AssignRoomPartition(ctx context.Context, roomID string, partition int32) (*model.BackfillRecord, bool)
  • func (r *BackfillRepo) ExpireRoomPartition(ctx context.Context, roomID string) error
  • func (r *BackfillRepo) GenerateFakePartition(ctx context.Context) int32
  • func (r *BackfillRepo) GetFinishedDomains(ctx context.Context, roomID string) (map[string]*BackfillJobFinishItem, bool)
  • func (r *BackfillRepo) GetUnfinishedRooms() ([]string, error)
  • func (r *BackfillRepo) InsertBackfillRecord(ctx context.Context, rec model.BackfillRecord) (bool, error)
  • func (r *BackfillRepo) IsBackfillFinished(ctx context.Context, roomID string) (isFinished, hasRec bool, err error)
  • func (r *BackfillRepo) LoadHistory() error
  • func (r *BackfillRepo) UnassignRoomPartition(ctx context.Context, roomID string) error
  • func (r *BackfillRepo) UpdateRecord(ctx context.Context, roomID string, depth int64, finished bool, ...) error
• type JoinRoomsData
• type JoinRoomsRepo
  • func NewJoinRoomsRepo(db model.FederationDatabase) *JoinRoomsRepo
  • func (r *JoinRoomsRepo) AddData(ctx context.Context, data *JoinRoomsData) error
  • func (r *JoinRoomsRepo) GetData(ctx context.Context, roomID string) (*JoinRoomsData, error)
  • func (r *JoinRoomsRepo) UpdateData(ctx context.Context, data *JoinRoomsData) error
• type JoinedRoomFinishedEventOffset
• type RecordItem
• type SendRecRepo
  • func NewSendRecRepo(db model.FederationDatabase, cache service.Cache) *SendRecRepo
  • func (r *SendRecRepo) AssignRoomPartition(ctx context.Context, roomID, domain string, partition int32) (*RecordItem, bool)
  • func (r *SendRecRepo) ExpireRoomPartition(ctx context.Context, roomID, domain string) error
  • func (r *SendRecRepo) GenerateFakePartition(ctx context.Context) int32
  • func (r *SendRecRepo) GetPenddingRooms(ctx context.Context, partition int32, count int) ([]string, []string)
  • func (r *SendRecRepo) GetRec(ctx context.Context, roomID, domain string) *RecordItem
  • func (r *SendRecRepo) IncrPendingSize(ctx context.Context, roomID, domain string, amt int, domainOffset int64) error
  • func (r *SendRecRepo) LoadRooms() error
  • func (r *SendRecRepo) ReleaseSendRec(ctx context.Context, roomID, domain string)
  • func (r *SendRecRepo) StoreRec(ctx context.Context, roomID, domain string, domainOffset int64) (*RecordItem, error)
  • func (r *SendRecRepo) UnassignRoomPartition(ctx context.Context, roomID, domain string) error
  • func (r *SendRecRepo) UpdateDomainOffset(ctx context.Context, roomID, domain, eventID string, domainOffset int64, ...)

Types

type BackfillJobFinishItem

type BackfillJobFinishItem struct {
	EventID      string
	StartEventID string
	StartOffset  int64
	Finished     bool
}

type BackfillRepo

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

func NewBackfillRepo

func NewBackfillRepo(
	db model.FederationDatabase,
	cache service.Cache,
) *BackfillRepo

func (*BackfillRepo) AssignRoomPartition

func (r *BackfillRepo) AssignRoomPartition(ctx context.Context, roomID string, partition int32) (*model.BackfillRecord, bool)

func (*BackfillRepo) ExpireRoomPartition

func (r *BackfillRepo) ExpireRoomPartition(ctx context.Context, roomID string) error

func (*BackfillRepo) GenerateFakePartition

func (r *BackfillRepo) GenerateFakePartition(ctx context.Context) int32

func (*BackfillRepo) GetFinishedDomains

func (r *BackfillRepo) GetFinishedDomains(ctx context.Context, roomID string) (map[string]*BackfillJobFinishItem, bool)

func (*BackfillRepo) GetUnfinishedRooms

func (r *BackfillRepo) GetUnfinishedRooms() ([]string, error)

func (*BackfillRepo) InsertBackfillRecord

func (r *BackfillRepo) InsertBackfillRecord(ctx context.Context, rec model.BackfillRecord) (bool, error)

func (*BackfillRepo) IsBackfillFinished

func (r *BackfillRepo) IsBackfillFinished(ctx context.Context, roomID string) (isFinished, hasRec bool, err error)

func (*BackfillRepo) LoadHistory

func (r *BackfillRepo) LoadHistory() error

func (*BackfillRepo) UnassignRoomPartition

func (r *BackfillRepo) UnassignRoomPartition(ctx context.Context, roomID string) error

func (*BackfillRepo) UpdateRecord

func (r *BackfillRepo) UpdateRecord(ctx context.Context, roomID string, depth int64, finished bool, finishedDomains string, states string) error

type JoinRoomsData

type JoinRoomsData struct {
	RoomID         string
	EventID        string
	RecvOffsets    string
	RecvOffsetsMap map[string]*JoinedRoomFinishedEventOffset
	HasJoined      bool
	// contains filtered or unexported fields
}

type JoinRoomsRepo

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

func NewJoinRoomsRepo

func NewJoinRoomsRepo(
	db model.FederationDatabase,
) *JoinRoomsRepo

func (*JoinRoomsRepo) AddData

func (r *JoinRoomsRepo) AddData(ctx context.Context, data *JoinRoomsData) error

func (*JoinRoomsRepo) GetData

func (r *JoinRoomsRepo) GetData(ctx context.Context, roomID string) (*JoinRoomsData, error)

func (*JoinRoomsRepo) UpdateData

func (r *JoinRoomsRepo) UpdateData(ctx context.Context, data *JoinRoomsData) error

type JoinedRoomFinishedEventOffset

type JoinedRoomFinishedEventOffset struct {
	EventID string
	Offset  int64
}

type RecordItem

type RecordItem struct {
	RoomID string
	Domain string

	EventID      string
	SendTimes    int32
	PendingSize  int32
	DomainOffset int64
}

type SendRecRepo

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

func NewSendRecRepo

func NewSendRecRepo(
	db model.FederationDatabase,
	cache service.Cache,
) *SendRecRepo

func (*SendRecRepo) AssignRoomPartition

func (r *SendRecRepo) AssignRoomPartition(ctx context.Context, roomID, domain string, partition int32) (*RecordItem, bool)

func (*SendRecRepo) ExpireRoomPartition

func (r *SendRecRepo) ExpireRoomPartition(ctx context.Context, roomID, domain string) error

func (*SendRecRepo) GenerateFakePartition

func (r *SendRecRepo) GenerateFakePartition(ctx context.Context) int32

func (*SendRecRepo) GetPenddingRooms

func (r *SendRecRepo) GetPenddingRooms(ctx context.Context, partition int32, count int) ([]string, []string)

func (*SendRecRepo) GetRec

func (r *SendRecRepo) GetRec(ctx context.Context, roomID, domain string) *RecordItem

func (*SendRecRepo) IncrPendingSize

func (r *SendRecRepo) IncrPendingSize(ctx context.Context, roomID, domain string, amt int, domainOffset int64) error

NOTE: domainOffset 只有在插入的时候才会update，所以这里填收到的事件的domain_offset即可

func (*SendRecRepo) LoadRooms

func (r *SendRecRepo) LoadRooms() error

func (*SendRecRepo) ReleaseSendRec

func (r *SendRecRepo) ReleaseSendRec(ctx context.Context, roomID, domain string)

func (*SendRecRepo) StoreRec

func (r *SendRecRepo) StoreRec(ctx context.Context, roomID, domain string, domainOffset int64) (*RecordItem, error)

func (*SendRecRepo) UnassignRoomPartition

func (r *SendRecRepo) UnassignRoomPartition(ctx context.Context, roomID, domain string) error

func (*SendRecRepo) UpdateDomainOffset

func (r *SendRecRepo) UpdateDomainOffset(ctx context.Context, roomID, domain, eventID string, domainOffset int64, pendingSizeDecr int32)