database

README

stablecog/sc-go/database

This hosts database-related items including Postgres, Redis, Supabase, and Meili.

Appendix

• ./ent - primary auto-generated files created by the ent ORM
• ./generate.sh - Needed to re-generate the files in ./ent when changes are made to the schema ./ent/schema
• ./repository - Collection of functions used to access and modify the SQL database.

Design Philosphy

The SQL database should not be accessed using the ent or supabase client directly. Our design philosphy is that database access should occur only within this package, and other packages should reference this package functions when interacting with the database.

This helps our codebase stay organized, testable, and reliable.

SQL interactions should happen within ./repository (excluding supabase), each file in repository should be associated with a schema in ./ent/schema. Excluding some special considerations such as stored procedures, direct access, etc.

Documentation

Index

• Variables
• func NewEntClient(connInfo SqlDBConn) (*ent.Client, error)
• type PendingCogRequestRedis
• type PostgresConn
  • func (c *PostgresConn) DSN() string
  • func (c *PostgresConn) Dialect() string
• type RedisStats
• type RedisWrapper
  • func NewRedis(ctx context.Context) (*RedisWrapper, error)
  • func (r *RedisWrapper) CacheEmbeddings(ctx context.Context, key string, embedding []float32) error
  • func (r *RedisWrapper) CacheTranslation(ctx context.Context, key string, translation string) error
  • func (r *RedisWrapper) DeleteCogRequestStreamID(ctx context.Context, requestID string) (int64, error)
  • func (r *RedisWrapper) DeleteDiscordToken(discordId string) error
  • func (r *RedisWrapper) EnqueueCogRequest(ctx context.Context, stream string, request interface{}) error
  • func (r *RedisWrapper) GetCogRequestStreamID(ctx context.Context, requestID string) (string, error)
  • func (r *RedisWrapper) GetDiscordTokenFromID(discordId string) (string, error)
  • func (r *RedisWrapper) GetEmbeddings(ctx context.Context, key string) ([]float32, error)
  • func (r *RedisWrapper) GetPendingGenerationAndUpscaleIDs(olderThan time.Duration) (generationOutputIDs, upscaleOutputIDs []PendingCogRequestRedis, err error)
  • func (r *RedisWrapper) GetQueueSize() (int64, error)
  • func (r *RedisWrapper) GetStatsCount() (stats *RedisStats, err error)
  • func (r *RedisWrapper) GetTranslation(ctx context.Context, key string) (string, error)
  • func (r *RedisWrapper) GetWorkerHealth() (shared.HEALTH_STATUS, error)
  • func (r *RedisWrapper) SetCogRequestStreamID(ctx context.Context, requestID string, streamID string) error
  • func (r *RedisWrapper) SetDiscordVerifyToken(discordId string) (string, error)
  • func (r *RedisWrapper) SetOutputCount(generationOutputCount, upscaleOutputCount, voiceoverOutputCount int) error
  • func (r *RedisWrapper) SetWorkerHealth(status shared.HEALTH_STATUS) error
  • func (r *RedisWrapper) XDelListOfIDs(ids []string) (deleted int64, err error)
• type SqlDBConn
  • func GetSqlDbConn(mock bool) (SqlDBConn, error)
• type SqliteConn
  • func (c *SqliteConn) DSN() string
  • func (c *SqliteConn) Dialect() string
• type SupabaseAuth
  • func NewSupabaseAuth() *SupabaseAuth
  • func (s *SupabaseAuth) AuthorizeWithDiscord() (authUrl string, err error)
  • func (s *SupabaseAuth) DeleteUser(id uuid.UUID) error
  • func (s *SupabaseAuth) GetSupabaseUserIdFromAccessToken(accessToken string) (id, email string, lastSignIn *time.Time, err error)

Variables

var SupabaseAuthUnauthorized = errors.New("Unauthorized")

Functions

func NewEntClient

func NewEntClient(connInfo SqlDBConn) (*ent.Client, error)

Types

type PendingCogRequestRedis

type PendingCogRequestRedis struct {
	RedisMsgid string
	Type       shared.ProcessType
	ID         uuid.UUID
}

type PostgresConn

type PostgresConn struct {
	Host     string
	Port     int
	Password string
	User     string
	DBName   string
}

func (*PostgresConn) DSN

func (c *PostgresConn) DSN() string

func (*PostgresConn) Dialect

func (c *PostgresConn) Dialect() string

type RedisStats

type RedisStats struct {
	GenerationOutputCount int `json:"generation_output_count"`
	UpscaleOutputCount    int `json:"upscale_output_count"`
	VoiceoverOutputCount  int `json:"voiceover_output_count"`
}

type RedisWrapper

type RedisWrapper struct {
	Client *redis.Client
	Ctx    context.Context
}

func NewRedis

func NewRedis(ctx context.Context) (*RedisWrapper, error)

Returns our *RedisWrapper, since we wrap some useful methods with the redis client

func (*RedisWrapper) CacheEmbeddings

func (r *RedisWrapper) CacheEmbeddings(ctx context.Context, key string, embedding []float32) error

Caching embeddings

func (*RedisWrapper) CacheTranslation

func (r *RedisWrapper) CacheTranslation(ctx context.Context, key string, translation string) error

Caching translations

func (*RedisWrapper) DeleteCogRequestStreamID

func (r *RedisWrapper) DeleteCogRequestStreamID(ctx context.Context, requestID string) (int64, error)

Delete the stream ID of the client for a given request ID

func (*RedisWrapper) DeleteDiscordToken

func (r *RedisWrapper) DeleteDiscordToken(discordId string) error

Delete token

func (*RedisWrapper) EnqueueCogRequest

func (r *RedisWrapper) EnqueueCogRequest(ctx context.Context, stream string, request interface{}) error

Enqueues a request to sc-worker

func (*RedisWrapper) GetCogRequestStreamID

func (r *RedisWrapper) GetCogRequestStreamID(ctx context.Context, requestID string) (string, error)

Get the stream ID of the client for a given request ID

func (*RedisWrapper) GetDiscordTokenFromID

func (r *RedisWrapper) GetDiscordTokenFromID(discordId string) (string, error)

Get discord ID from token

func (*RedisWrapper) GetEmbeddings

func (r *RedisWrapper) GetEmbeddings(ctx context.Context, key string) ([]float32, error)

Retrieve from cache

func (*RedisWrapper) GetPendingGenerationAndUpscaleIDs

func (r *RedisWrapper) GetPendingGenerationAndUpscaleIDs(olderThan time.Duration) (generationOutputIDs, upscaleOutputIDs []PendingCogRequestRedis, err error)

Get pending request IDs on queue that are stale olderThan will be subtracted from the current time to return requests older than that

func (*RedisWrapper) GetQueueSize

func (r *RedisWrapper) GetQueueSize() (int64, error)

func (*RedisWrapper) GetStatsCount

func (r *RedisWrapper) GetStatsCount() (stats *RedisStats, err error)

Get generate and upscale count stats

func (*RedisWrapper) GetTranslation

func (r *RedisWrapper) GetTranslation(ctx context.Context, key string) (string, error)

Retrieve from cache

func (*RedisWrapper) GetWorkerHealth

func (r *RedisWrapper) GetWorkerHealth() (shared.HEALTH_STATUS, error)

func (*RedisWrapper) SetCogRequestStreamID

func (r *RedisWrapper) SetCogRequestStreamID(ctx context.Context, requestID string, streamID string) error

Keep track of request ID to cog, with stream ID of the client, for timeout tracking

func (*RedisWrapper) SetDiscordVerifyToken

func (r *RedisWrapper) SetDiscordVerifyToken(discordId string) (string, error)

Set verify token and return token

func (*RedisWrapper) SetOutputCount

func (r *RedisWrapper) SetOutputCount(generationOutputCount, upscaleOutputCount, voiceoverOutputCount int) error

Set generate and upscale count stats

func (*RedisWrapper) SetWorkerHealth

func (r *RedisWrapper) SetWorkerHealth(status shared.HEALTH_STATUS) error

func (*RedisWrapper) XDelListOfIDs

func (r *RedisWrapper) XDelListOfIDs(ids []string) (deleted int64, err error)

type SqlDBConn

type SqlDBConn interface {
	DSN() string
	Dialect() string
}

func GetSqlDbConn

func GetSqlDbConn(mock bool) (SqlDBConn, error)

Gets the DB connection information based on environment variables

type SqliteConn

type SqliteConn struct {
	FileName string
	Mode     string
}

func (*SqliteConn) DSN

func (c *SqliteConn) DSN() string

func (*SqliteConn) Dialect

func (c *SqliteConn) Dialect() string

type SupabaseAuth

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

func NewSupabaseAuth

func NewSupabaseAuth() *SupabaseAuth

Returns gotrue client with keys

func (*SupabaseAuth) AuthorizeWithDiscord

func (s *SupabaseAuth) AuthorizeWithDiscord() (authUrl string, err error)

func (*SupabaseAuth) DeleteUser

func (s *SupabaseAuth) DeleteUser(id uuid.UUID) error

func (*SupabaseAuth) GetSupabaseUserIdFromAccessToken

func (s *SupabaseAuth) GetSupabaseUserIdFromAccessToken(accessToken string) (id, email string, lastSignIn *time.Time, err error)