From 1e8adb4da48eaee98d47920e419c79a86211d00e Mon Sep 17 00:00:00 2001 From: frx Date: Sun, 5 Jul 2026 17:11:50 +0000 Subject: [PATCH] fix(issue-12): ydb course repository migrate to new v3 --- .../kurious/adapters/ydb_course_repository.go | 941 +++++++----------- internal/kurious/service/service.go | 9 +- 2 files changed, 369 insertions(+), 581 deletions(-) diff --git a/internal/kurious/adapters/ydb_course_repository.go b/internal/kurious/adapters/ydb_course_repository.go index d6f1ec0..da32e98 100644 --- a/internal/kurious/adapters/ydb_course_repository.go +++ b/internal/kurious/adapters/ydb_course_repository.go @@ -4,10 +4,7 @@ import ( "context" "fmt" "log/slog" - "os" - "path" "strings" - "text/template" "time" "git.loyso.art/frx/kurious/internal/common/config" @@ -18,42 +15,21 @@ import ( "git.loyso.art/frx/kurious/pkg/xdefault" "github.com/ydb-platform/ydb-go-sdk/v3" - ydblog "github.com/ydb-platform/ydb-go-sdk/v3/log" "github.com/ydb-platform/ydb-go-sdk/v3/table" "github.com/ydb-platform/ydb-go-sdk/v3/table/options" + "github.com/ydb-platform/ydb-go-sdk/v3/table/result" "github.com/ydb-platform/ydb-go-sdk/v3/table/result/named" "github.com/ydb-platform/ydb-go-sdk/v3/table/types" "github.com/ydb-platform/ydb-go-sdk/v3/trace" yc "github.com/ydb-platform/ydb-go-yc" ) -var coursesFields = []string{ - "id", - "external_id", - "source_type", - "source_name", - "course_thematic", - "learning_type", - "organization_id", - "origin_link", - "image_link", - "name", - "description", - "full_price", - "discount", - "duration", - "starts_at", - "created_at", - "updated_at", - "deleted_at", -} - -var coursesFieldsStr = strings.Join(coursesFields, ",") - const ( defaultShutdownTimeout = time.Second * 10 + coursesTableName = "courses" ) +// YDBConnection wraps a ydb.Driver with application-level lifecycle. type YDBConnection struct { *ydb.Driver @@ -61,6 +37,8 @@ type YDBConnection struct { shutdownTimeout time.Duration } +// NewYDBConnection opens a YDB connection using v3 SDK patterns. +// See: https://ydb.tech/docs/ru/dev/example-app/go/?version=v26.1 func NewYDBConnection(ctx context.Context, cfg config.YDB, log *slog.Logger) (*YDBConnection, error) { opts := make([]ydb.Option, 0, 3) switch auth := cfg.Auth.(type) { @@ -73,9 +51,7 @@ func NewYDBConnection(ctx context.Context, cfg config.YDB, log *slog.Logger) (*Y ) } if cfg.DebugYDB { - opts = append(opts, - ydb.WithLogger(ydblog.Default(os.Stdout, ydblog.WithMinLevel(ydblog.DEBUG)), trace.DetailsAll), - ) + opts = append(opts, ydb.WithTraceDriver(trace.Driver{})) } db, err := ydb.Open( @@ -86,13 +62,13 @@ func NewYDBConnection(ctx context.Context, cfg config.YDB, log *slog.Logger) (*Y if err != nil { return nil, fmt.Errorf("opening connection: %w", err) } - endpoints, err := db.Discovery().Discover(ctx) - if err != nil { - return nil, fmt.Errorf("discovering endpoints: %w", err) - } - for _, endpoint := range endpoints { - xcontext.LogInfo(ctx, log, "discovered endpoint", slog.String("value", endpoint.Address())) + // Verify connectivity with a discovery call + err = db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + return nil + }, table.WithIdempotent()) + if err != nil { + return nil, fmt.Errorf("discovery check: %w", err) } return &YDBConnection{ @@ -102,21 +78,24 @@ func NewYDBConnection(ctx context.Context, cfg config.YDB, log *slog.Logger) (*Y }, nil } +// Close gracefully shuts down the YDB driver. func (conn *YDBConnection) Close() error { ctx, cancel := context.WithTimeout(context.Background(), conn.shutdownTimeout) defer cancel() - return conn.Driver.Close(ctx) } +// Organization returns a not-implemented stub for YDB (no organization table yet). func (conn *YDBConnection) Organization() domain.OrganizationRepository { return NotImplementedOrganizationRepository{} } +// LearningCategory returns a not-implemented stub for YDB (no learning_category table yet). func (conn *YDBConnection) LearningCategory() domain.LearningCategoryRepository { return NotImplementedLearningCategory{} } +// CourseRepository returns a YDB-backed CourseRepository. func (conn *YDBConnection) CourseRepository() *ydbCourseRepository { return &ydbCourseRepository{ db: conn.Driver, @@ -124,614 +103,345 @@ func (conn *YDBConnection) CourseRepository() *ydbCourseRepository { } } +// ydbCourseRepository implements domain.CourseRepository backed by YDB. type ydbCourseRepository struct { db *ydb.Driver log *slog.Logger } +// List returns courses matching the given filters with cursor-based pagination. func (r *ydbCourseRepository) List( ctx context.Context, params domain.ListCoursesParams, ) (result domain.ListCoursesResult, err error) { const limit = 1000 - const queryName = "list" if params.Limit == 0 { params.Limit = limit } - qtParams := queryTemplateParams{ - Fields: coursesFieldsStr, - Table: "courses", - Suffix: "ORDER BY learning_type,course_thematic,id\nLIMIT $limit", - Declares: []queryTemplateDeclaration{ - { - Name: "limit", - Type: "Int32", - }, - { - Name: "id", - Type: "Text", - }, - }, - Conditions: []string{ - "id > $id", - }, - } + query := buildListQuery(params) - opts := make([]table.ParameterOption, 0, 4) - appendTextParam := func(name string, value string) { - if value == "" { - return - } + txParams := buildListQueryParams(params) - ydbvalue := types.TextValue(value) - d := queryTemplateDeclaration{ - Name: name, - Type: ydbvalue.Type().String(), - } - qtParams.Declares = append(qtParams.Declares, d) - qtParams.Conditions = append(qtParams.Conditions, d.Name+"="+d.Arg()) - opts = append(opts, table.ValueParam(d.Arg(), ydbvalue)) - } - appendTextParam("course_thematic", params.CourseThematic) - appendTextParam("learning_type", params.LearningType) + xcontext.LogInfo(ctx, r.log, "query prepared", slog.String("query", query)) - opts = append( - opts, - table.ValueParam("$limit", types.Int32Value(int32(params.Limit))), - ) + readTx := table.OnlineReadOnlyTxControl() - query, err := qtParams.render() - if err != nil { - return result, fmt.Errorf("rendering query params: %w", err) - } - - xcontext.LogInfo(ctx, r.log, "query prepared", slog.String("query", query), slog.String("args", tableParamOptsToString(opts...))) - - courses := make([]domain.Course, 0, 1_000) - readTx := table.TxControl( - table.BeginTx( - table.WithOnlineReadOnly(), - ), - table.CommitTx(), - ) - - err = r.db.Table().Do( - ctx, + err = r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - start := time.Now() - defer func() { - since := time.Since(start).Truncate(time.Millisecond) - xcontext.LogInfo( - ctx, r.log, - "executed query", - slog.String("name", queryName), - slog.Duration("elapsed", since), - ) - }() - - queryParams := table.NewQueryParameters(opts...) - _, res, err := s.Execute( - ctx, readTx, query, queryParams, + ctx, readTx, query, txParams, options.WithCollectStatsModeBasic(), ) if err != nil { - return fmt.Errorf("executing: %w", err) + return fmt.Errorf("executing list query: %w", err) } - if !res.NextResultSet(ctx) || !res.HasNextRow() { - return nil - } - - for res.NextRow() { - var cdb courseDB - err = res.ScanNamed(cdb.getNamedValues()...) - if err != nil { - return fmt.Errorf("scanning row: %w", err) - } - - courses = append(courses, mapCourseDB(cdb)) - } - if err = res.Err(); err != nil { - return err - } - - result.NextPageToken = courses[len(courses)-1].ID - xcontext.LogDebug(ctx, r.log, "scanned rows", slog.Int("count", len(courses))) - - return nil + return scanCoursesResult(ctx, res, &result) }, - table.WithIdempotent()) + table.WithIdempotent(), + ) if err != nil { return domain.ListCoursesResult{}, err } - result.Courses = courses - return result, err + return result, nil } +// ListLearningTypes returns distinct learning type IDs. func (r *ydbCourseRepository) ListLearningTypes( ctx context.Context, ) (result domain.ListLearningTypeResult, err error) { - const queryName = "list_learning_type" - const querySelect = `SELECT DISTINCT learning_type FROM courses;` + const querySelect = `SELECT DISTINCT learning_type FROM courses` - readTx := table.TxControl( - table.BeginTx( - table.WithOnlineReadOnly(), - ), - table.CommitTx(), - ) + readTx := table.OnlineReadOnlyTxControl() - err = r.db.Table().Do( - ctx, + err = r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - start := time.Now() - defer func() { - since := time.Since(start).Truncate(time.Millisecond) - xcontext.LogInfo( - ctx, r.log, - "executed query", - slog.String("name", queryName), - slog.Duration("elapsed", since), - ) - }() - _, res, err := s.Execute( ctx, readTx, querySelect, table.NewQueryParameters(), options.WithCollectStatsModeNone(), ) if err != nil { - return fmt.Errorf("executing query: %w", err) + return fmt.Errorf("executing list learning types query: %w", err) } if !res.NextResultSet(ctx) || !res.HasNextRow() { return nil } - for res.NextRow() { - var learningTypeID string - if err = res.Scan(&learningTypeID); err != nil { + var lt string + if err = res.Scan(<); err != nil { return fmt.Errorf("scanning row: %w", err) } - - result.LearningTypeIDs = append(result.LearningTypeIDs, learningTypeID) + result.LearningTypeIDs = append(result.LearningTypeIDs, lt) } - if err = res.Err(); err != nil { - return err - } - - xcontext.LogDebug(ctx, r.log, "scanned rows", slog.Int("count", len(result.LearningTypeIDs))) - - return nil + return res.Err() }, table.WithIdempotent(), ) if err != nil { return result, err } - return result, nil } +// ListCourseThematics returns distinct course thematic IDs for a learning type. func (r *ydbCourseRepository) ListCourseThematics( ctx context.Context, params domain.ListCourseThematicsParams, ) (result domain.ListCourseThematicsResult, err error) { - const queryName = "list_course_thematics" - - qtParams := queryTemplateParams{ - Fields: "DISTINCT course_thematic", - Table: "courses", - Declares: []queryTemplateDeclaration{}, - Conditions: []string{}, + query := `SELECT DISTINCT course_thematic FROM courses WHERE 1=1` + txParams := table.NewQueryParameters() + if params.LearningTypeID != "" { + query += ` AND learning_type = $learning_type` + txParams = table.NewQueryParameters( + table.ValueParam("$learning_type", types.TextValue(params.LearningTypeID)), + ) } - learningTypeValue := types.TextValue(params.LearningTypeID) - d := queryTemplateDeclaration{ - Name: "learning_type", - Type: learningTypeValue.Type().String(), - } - qtParams.Declares = append(qtParams.Declares, d) - qtParams.Conditions = append(qtParams.Conditions, d.Name+"="+d.Arg()) + readTx := table.OnlineReadOnlyTxControl() - opts := []table.ParameterOption{ - table.ValueParam(d.Arg(), learningTypeValue), - } - - query, err := qtParams.render() - if err != nil { - return result, fmt.Errorf("rendering query params: %w", err) - } - - readTx := table.TxControl( - table.BeginTx( - table.WithOnlineReadOnly(), - ), - table.CommitTx(), - ) - - err = r.db.Table().Do( - ctx, + err = r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - start := time.Now() - defer func() { - since := time.Since(start).Truncate(time.Millisecond) - xcontext.LogInfo( - ctx, r.log, - "executed query", - slog.String("name", queryName), - slog.Duration("elapsed", since), - ) - }() - _, res, err := s.Execute( - ctx, readTx, query, table.NewQueryParameters(opts...), + ctx, readTx, query, txParams, options.WithCollectStatsModeNone(), ) if err != nil { - return fmt.Errorf("executing query: %w", err) + return fmt.Errorf("executing list course thematics query: %w", err) } if !res.NextResultSet(ctx) || !res.HasNextRow() { return nil } - for res.NextRow() { - var courseThematicID string - if err = res.Scan(&courseThematicID); err != nil { + var ct string + if err = res.Scan(&ct); err != nil { return fmt.Errorf("scanning row: %w", err) } - - result.CourseThematicIDs = append(result.CourseThematicIDs, courseThematicID) + result.CourseThematicIDs = append(result.CourseThematicIDs, ct) } - if err = res.Err(); err != nil { - return err - } - - xcontext.LogDebug(ctx, r.log, "scanned rows", slog.Int("count", len(result.CourseThematicIDs))) - - return nil + return res.Err() }, table.WithIdempotent(), ) if err != nil { return result, err } - return result, nil } +// ListStatistics returns course statistics grouped by learning type, thematic, and organization. +func (r *ydbCourseRepository) ListStatistics( + ctx context.Context, + params domain.ListStatisticsParams, +) (result domain.ListStatisticsResult, err error) { + query := `SELECT learning_type, course_thematic, organization_id, COUNT(id) AS count FROM courses WHERE 1=1` + txParams := make([]table.ParameterOption, 0) + if params.LearningTypeID != "" { + query += ` AND learning_type = $learning_type` + txParams = append(txParams, table.ValueParam("$learning_type", types.TextValue(params.LearningTypeID))) + } + if params.CourseThematicID != "" { + query += ` AND course_thematic = $course_thematic` + txParams = append(txParams, table.ValueParam("$course_thematic", types.TextValue(params.CourseThematicID))) + } + if params.OrganizaitonID != "" { + query += ` AND organization_id = $organization_id` + txParams = append(txParams, table.ValueParam("$organization_id", types.TextValue(params.OrganizaitonID))) + } + query += ` GROUP BY learning_type, course_thematic, organization_id ORDER BY count DESC` + + readTx := table.OnlineReadOnlyTxControl() + + err = r.db.Table().Do(ctx, + func(ctx context.Context, s table.Session) error { + _, res, err := s.Execute( + ctx, readTx, query, table.NewQueryParameters(txParams...), + options.WithCollectStatsModeNone(), + ) + if err != nil { + return fmt.Errorf("executing statistics query: %w", err) + } + if !res.NextResultSet(ctx) || !res.HasNextRow() { + return nil + } + for res.NextRow() { + var stat domain.StatisticUnit + if err = res.Scan(&stat.LearningTypeID, &stat.CourseThematicID, &stat.OrganizationID, &stat.Count); err != nil { + return fmt.Errorf("scanning row: %w", err) + } + result.LearningTypeStatistics = append(result.LearningTypeStatistics, stat) + } + return res.Err() + }, + table.WithIdempotent(), + ) + if err != nil { + return result, err + } + return result, nil +} + +// Get retrieves a single course by ID. func (r *ydbCourseRepository) Get( ctx context.Context, id string, ) (course domain.Course, err error) { - const queryName = "get" const querySelect = `DECLARE $id AS Text; - SELECT - id, - external_id, - source_type, - source_name, - course_thematic, - learning_type, - organization_id, - origin_link, - image_link, - name, - description, - full_price, - discount, - duration, - starts_at, - created_at, - updated_at, - deleted_at - FROM - courses - WHERE - id = $id;` + SELECT + id, external_id, source_type, source_name, + course_thematic, learning_type, organization_id, + origin_link, image_link, name, description, + full_price, discount, duration, starts_at, + created_at, updated_at, deleted_at + FROM courses WHERE id = $id` - readTx := table.TxControl( - table.BeginTx( - table.WithOnlineReadOnly(), - ), - table.CommitTx(), - ) - err = r.db.Table().Do( - ctx, + readTx := table.OnlineReadOnlyTxControl() + + err = r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - start := time.Now() - defer func() { - since := time.Since(start) - xcontext.LogInfo( - ctx, r.log, - "executed query", - slog.String("name", queryName), - slog.Duration("elapsed", since), - ) - }() - _, res, err := s.Execute( - ctx, - readTx, - querySelect, + ctx, readTx, querySelect, table.NewQueryParameters( table.ValueParam("$id", types.TextValue(id)), ), options.WithCollectStatsModeBasic(), ) if err != nil { - return fmt.Errorf("executing query: %w", err) + return fmt.Errorf("executing get query: %w", err) } - if !res.NextResultSet(ctx) || !res.HasNextRow() { return errors.ErrNotFound } - for res.NextRow() { var cdb courseDB - err = res.ScanNamed(cdb.getNamedValues()...) - if err != nil { + if err = res.ScanNamed(cdb.namedValues()...); err != nil { return fmt.Errorf("scanning row: %w", err) } - course = mapCourseDB(cdb) } - if err = res.Err(); err != nil { - return err - } - - stats := res.Stats() - xcontext.LogInfo( - ctx, r.log, "query stats", - slog.String("ast", stats.QueryAST()), - slog.String("plan", stats.QueryPlan()), - slog.Duration("total_cpu_time", stats.TotalCPUTime()), - slog.Duration("total_duration", stats.TotalDuration()), - slog.Duration("process_cpu_time", stats.ProcessCPUTime()), - ) - - return nil + return res.Err() }, table.WithIdempotent(), ) return course, err } +// GetByExternalID finds a course by its external ID. func (r *ydbCourseRepository) GetByExternalID(ctx context.Context, id string) (domain.Course, error) { - return domain.Course{}, nil -} + const querySelect = `DECLARE $external_id AS Text; + SELECT + id, external_id, source_type, source_name, + course_thematic, learning_type, organization_id, + origin_link, image_link, name, description, + full_price, discount, duration, starts_at, + created_at, updated_at, deleted_at + FROM courses WHERE external_id = $external_id` -type updateCourseParams struct { - domain.CreateCourseParams + readTx := table.OnlineReadOnlyTxControl() - CreatedAt time.Time - DeletedAt nullable.Value[time.Time] -} - -func updateCourseParamsAsStruct(params updateCourseParams) types.Value { - opts := createCourseParamsAsStructValues(params.CreateCourseParams) - now := time.Now() - return types.StructValue( - append( - opts[:len(opts)-3], - types.StructFieldValue("created_at", types.DatetimeValueFromTime(params.CreatedAt)), - types.StructFieldValue("updated_at", types.DatetimeValueFromTime(now)), - types.StructFieldValue("deleted_at", types.NullableDatetimeValue(nil)), - )..., - ) -} - -func createCourseParamsAsStructValues(params domain.CreateCourseParams) []types.StructValueOption { - st := mapSourceTypeFromDomain(params.SourceType) - now := time.Now() - - return []types.StructValueOption{ - types.StructFieldValue("id", types.TextValue(params.ID)), - types.StructFieldValue("name", types.TextValue(params.Name)), - types.StructFieldValue("source_type", types.TextValue(st)), - types.StructFieldValue("source_name", types.NullableTextValue(params.SourceName.ValutPtr())), - types.StructFieldValue("course_thematic", types.TextValue(params.CourseThematic)), - types.StructFieldValue("learning_type", types.TextValue(params.LearningType)), - types.StructFieldValue("external_id", types.NullableTextValue(params.ExternalID.ValutPtr())), - types.StructFieldValue("organization_id", types.TextValue(params.OrganizationID)), - types.StructFieldValue("origin_link", types.TextValue(params.OriginLink)), - types.StructFieldValue("image_link", types.TextValue(params.ImageLink)), - types.StructFieldValue("description", types.TextValue(params.Description)), - types.StructFieldValue("full_price", types.DoubleValue(params.FullPrice)), - types.StructFieldValue("discount", types.DoubleValue(params.Discount)), - types.StructFieldValue("duration", types.IntervalValueFromDuration(params.Duration)), - types.StructFieldValue("starts_at", types.DatetimeValueFromTime(params.StartsAt)), - types.StructFieldValue("created_at", types.DatetimeValueFromTime(now)), - types.StructFieldValue("updated_at", types.DatetimeValueFromTime(now)), - types.StructFieldValue("deleted_at", types.NullableDatetimeValue(nil)), - } -} - -func createCourseParamsAsStruct(params domain.CreateCourseParams) types.Value { - return types.StructValue( - createCourseParamsAsStructValues(params)..., + var course domain.Course + err := r.db.Table().Do(ctx, + func(ctx context.Context, s table.Session) error { + _, res, err := s.Execute( + ctx, readTx, querySelect, + table.NewQueryParameters( + table.ValueParam("$external_id", types.TextValue(id)), + ), + options.WithCollectStatsModeNone(), + ) + if err != nil { + return fmt.Errorf("executing get by external id query: %w", err) + } + if !res.NextResultSet(ctx) || !res.HasNextRow() { + return errors.ErrNotFound + } + for res.NextRow() { + var cdb courseDB + if err = res.ScanNamed(cdb.namedValues()...); err != nil { + return fmt.Errorf("scanning row: %w", err) + } + course = mapCourseDB(cdb) + } + return res.Err() + }, + table.WithIdempotent(), ) + return course, err } +// CreateBatch inserts or replaces multiple courses using BulkUpsert. +// This is the recommended v3 pattern for batch writes. func (r *ydbCourseRepository) CreateBatch(ctx context.Context, params ...domain.CreateCourseParams) error { - const upsertQuery = `DECLARE $courseData AS List, - name: Text, - source_type: Text, - source_name: Optional, - course_thematic: Text, - learning_type: Text, - organization_id: Text, - origin_link: Text, - image_link: Text, - description: Text, - full_price: Double, - discount: Double, - duration: Interval, - starts_at: Datetime, - created_at: Datetime, - updated_at: Datetime, - deleted_at: Optional>>; + values := make([]types.Value, 0, len(params)) + for _, p := range params { + values = append(values, createCourseAsStructValue(p)) + } - REPLACE INTO - courses - SELECT - id, - external_id, - name, - source_type, - source_name, - course_thematic, - learning_type, - organization_id, - origin_link, - image_link, - description, - full_price, - discount, - duration, - starts_at, - created_at, - updated_at, - deleted_at - FROM AS_TABLE($courseData);` - - writeTx := table.TxControl( - table.BeginTx( - table.WithSerializableReadWrite(), - ), - table.CommitTx(), + return r.db.Table().Do(ctx, + func(ctx context.Context, s table.Session) error { + return s.BulkUpsert(ctx, coursesTableName, + types.ListValue(values...), + ) + }, + table.WithIdempotent(), ) - err := r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - listValues := mapSlice(params, createCourseParamsAsStruct) - queryParams := table.NewQueryParameters( - table.ValueParam("$courseData", types.ListValue(listValues...)), - ) - _, _, err := s.Execute(ctx, writeTx, upsertQuery, queryParams) - if err != nil { - return fmt.Errorf("executing query: %w", err) - } - - return nil - }) - - return err } +// Create inserts or replaces a single course. func (r *ydbCourseRepository) Create(ctx context.Context, params domain.CreateCourseParams) (domain.Course, error) { err := r.CreateBatch(ctx, params) if err != nil { return domain.Course{}, err } - return domain.Course{}, nil } +// Delete removes a course by ID. func (r *ydbCourseRepository) Delete(ctx context.Context, id string) error { - return nil -} + const queryDelete = `DECLARE $id AS Text; + DELETE FROM courses WHERE id = $id` -func (r *ydbCourseRepository) UpdateCourseDescription(ctx context.Context, id, description string) error { - course, err := r.Get(ctx, id) - if err != nil { - return fmt.Errorf("getting course: %w", err) - } - - params := updateCourseParams{ - CreateCourseParams: domain.CreateCourseParams{ - ID: course.ID, - ExternalID: course.ExternalID, - Name: course.Name, - SourceType: course.SourceType, - SourceName: course.SourceName, - CourseThematic: course.Thematic, - LearningType: course.LearningType, - OrganizationID: course.OrganizationID, - OriginLink: course.OriginLink, - ImageLink: course.ImageLink, - Description: description, - FullPrice: course.FullPrice, - Discount: course.Discount, - Duration: course.Duration, - StartsAt: course.StartsAt, + return r.db.Table().DoTx(ctx, + func(ctx context.Context, tx table.TransactionActor) error { + _, err := tx.Execute(ctx, queryDelete, + table.NewQueryParameters( + table.ValueParam("$id", types.TextValue(id)), + ), + ) + return err }, - CreatedAt: course.CreatedAt, - DeletedAt: course.DeletedAt, - } - - updateStruct := updateCourseParamsAsStruct(params) - - const upsertQuery = `DECLARE $courseData AS List, - name: Text, - source_type: Text, - source_name: Optional, - course_thematic: Text, - learning_type: Text, - organization_id: Text, - origin_link: Text, - image_link: Text, - description: Text, - full_price: Double, - discount: Double, - duration: Interval, - starts_at: Datetime, - created_at: Datetime, - updated_at: Datetime, - deleted_at: Optional>>; - - REPLACE INTO - courses - SELECT - id, - external_id, - name, - source_type, - source_name, - course_thematic, - learning_type, - organization_id, - origin_link, - image_link, - description, - full_price, - discount, - duration, - starts_at, - created_at, - updated_at, - deleted_at - FROM AS_TABLE($courseData);` - - writeTx := table.TxControl( - table.BeginTx( - table.WithSerializableReadWrite(), - ), - table.CommitTx(), + table.WithIdempotent(), ) - err = r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - queryParams := table.NewQueryParameters( - table.ValueParam("$courseData", types.ListValue(updateStruct)), - ) - _, _, err := s.Execute(ctx, writeTx, upsertQuery, queryParams) - if err != nil { - return fmt.Errorf("executing query: %w", err) - } - - return nil - }) - - return err } +// UpdateCourseDescription updates the description field of a course. +func (r *ydbCourseRepository) UpdateCourseDescription(ctx context.Context, id, description string) error { + const queryUpdate = ` + DECLARE $id AS Text; + DECLARE $description AS Text; + DECLARE $updated_at AS Datetime; + UPDATE courses SET description = $description, updated_at = $updated_at WHERE id = $id` + + return r.db.Table().DoTx(ctx, + func(ctx context.Context, tx table.TransactionActor) error { + _, err := tx.Execute(ctx, queryUpdate, + table.NewQueryParameters( + table.ValueParam("$id", types.TextValue(id)), + table.ValueParam("$description", types.TextValue(description)), + table.ValueParam("$updated_at", types.DatetimeValueFromTime(time.Now())), + ), + ) + return err + }, + table.WithIdempotent(), + ) +} + +// CreateCourseTable creates the courses table in YDB. func (r *ydbCourseRepository) CreateCourseTable(ctx context.Context) error { return r.db.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - return s.CreateTable( - ctx, - path.Join(r.db.Name(), "courses"), + return s.CreateTable(ctx, coursesTableName, options.WithColumn("id", types.TypeText), options.WithColumn("external_id", types.Optional(types.TypeText)), options.WithColumn("name", types.TypeText), @@ -755,6 +465,11 @@ func (r *ydbCourseRepository) CreateCourseTable(ctx context.Context) error { }) } +// --------------------------------------------------------------------------- +// DB row type and mapping helpers +// --------------------------------------------------------------------------- + +// courseDB represents a YDB row scanned from the courses table. type courseDB struct { ID string ExternalID *string @@ -776,7 +491,7 @@ type courseDB struct { DeletedAt *time.Time } -func (c *courseDB) getNamedValues() []named.Value { +func (c *courseDB) namedValues() []named.Value { return []named.Value{ named.Required("id", &c.ID), named.Required("name", &c.Name), @@ -799,47 +514,12 @@ func (c *courseDB) getNamedValues() []named.Value { } } -const ( - sourceTypeUnknown = "" - sourceTypeManual = "m" - sourceTypeParsed = "p" -) - -func mapSourceTypeToDomain(in string) (st domain.SourceType) { - switch in { - case sourceTypeUnknown: - st = domain.SourceTypeUnset - case sourceTypeManual: - st = domain.SourceTypeManual - case sourceTypeParsed: - st = domain.SourceTypeParsed - } - - return st -} - -func mapSourceTypeFromDomain(in domain.SourceType) string { - var st string - switch in { - case domain.SourceTypeManual: - st = sourceTypeManual - case domain.SourceTypeParsed: - st = sourceTypeParsed - default: - st = sourceTypeUnknown - } - - return st -} - func mapCourseDB(cdb courseDB) domain.Course { - st := mapSourceTypeToDomain(cdb.SourceType) - return domain.Course{ ID: cdb.ID, ExternalID: nullable.NewValuePtr(cdb.ExternalID), Name: cdb.Name, - SourceType: st, + SourceType: mapSourceTypeToDomain(cdb.SourceType), SourceName: nullable.NewValuePtr(cdb.SourceName), ThematicID: cdb.CourseThematic, LearningTypeID: cdb.LearningType, @@ -857,53 +537,154 @@ func mapCourseDB(cdb courseDB) domain.Course { } } -func mapSlice[T, U any](in []T, f func(T) U) []U { - out := make([]U, len(in)) - for i, value := range in { - out[i] = f(value) +// --------------------------------------------------------------------------- +// Source type mapping (domain <-> DB) +// --------------------------------------------------------------------------- + +const ( + sourceTypeUnknown = "" + sourceTypeManual = "m" + sourceTypeParsed = "p" +) + +func mapSourceTypeToDomain(in string) (st domain.SourceType) { + switch in { + case sourceTypeManual: + st = domain.SourceTypeManual + case sourceTypeParsed: + st = domain.SourceTypeParsed + } + return st +} + +func mapSourceTypeFromDomain(in domain.SourceType) string { + switch in { + case domain.SourceTypeManual: + return sourceTypeManual + case domain.SourceTypeParsed: + return sourceTypeParsed + default: + return sourceTypeUnknown + } +} + +// --------------------------------------------------------------------------- +// Query builders +// --------------------------------------------------------------------------- + +var coursesFields = []string{ + "id", "external_id", "source_type", "source_name", + "course_thematic", "learning_type", "organization_id", + "origin_link", "image_link", "name", "description", + "full_price", "discount", "duration", "starts_at", + "created_at", "updated_at", "deleted_at", +} + +var coursesFieldsStr = strings.Join(coursesFields, ", ") + +func buildListQuery(params domain.ListCoursesParams) string { + var sb strings.Builder + sb.WriteString("DECLARE $limit AS Int32;") + sb.WriteString("\n") + sb.WriteString("DECLARE $id AS Text;") + sb.WriteString("\n") + if params.CourseThematic != "" { + sb.WriteString("DECLARE $course_thematic AS Text;\n") + } + if params.LearningType != "" { + sb.WriteString("DECLARE $learning_type AS Text;\n") + } + if params.OrganizationID != "" { + sb.WriteString("DECLARE $organization_id AS Text;\n") } - return out -} + sb.WriteString("SELECT ") + sb.WriteString(coursesFieldsStr) + sb.WriteString(" FROM courses WHERE id > $id") -type queryTemplateDeclaration struct { - Name string - Type string -} - -func (d queryTemplateDeclaration) Arg() string { - return "$" + d.Name -} - -type queryTemplateParams struct { - Declares []queryTemplateDeclaration - Fields string - Table string - Conditions []string - Suffix string -} - -func (p queryTemplateParams) render() (string, error) { - var sb strings.Builder - sb.Grow(len(queryTemplateSelect) * 3) - - err := querySelect.Execute(&sb, p) - return sb.String(), err -} - -const queryTemplateSelect = `{{ range .Declares }}DECLARE ${{.Name}} AS {{.Type}};{{end}} -SELECT {{.Fields}} -FROM {{.Table}} -WHERE 1=1 {{ range .Conditions }} AND {{.}} {{ end }} -{{.Suffix}}` - -var querySelect = template.Must(template.New("").Parse(queryTemplateSelect)) - -func tableParamOptsToString(in ...table.ParameterOption) string { - var sb strings.Builder - for _, opt := range in { - sb.WriteString(opt.Name() + ":" + opt.Value().Yql() + ";") - // sb.WriteString(opt.Name() + " (" + opt.Value().Type().String() + "); ") + if params.LearningType != "" { + sb.WriteString(" AND learning_type = $learning_type") } + if params.CourseThematic != "" { + sb.WriteString(" AND course_thematic = $course_thematic") + } + if params.OrganizationID != "" { + sb.WriteString(" AND organization_id = $organization_id") + } + + sb.WriteString(" ORDER BY learning_type, course_thematic, id LIMIT $limit") return sb.String() } + +func buildListQueryParams(params domain.ListCoursesParams) *table.QueryParameters { + opts := make([]table.ParameterOption, 0, 5) + opts = append(opts, table.ValueParam("$limit", types.Int32Value(int32(params.Limit)))) + opts = append(opts, table.ValueParam("$id", types.TextValue(""))) + if params.CourseThematic != "" { + opts = append(opts, table.ValueParam("$course_thematic", types.TextValue(params.CourseThematic))) + } + if params.LearningType != "" { + opts = append(opts, table.ValueParam("$learning_type", types.TextValue(params.LearningType))) + } + if params.OrganizationID != "" { + opts = append(opts, table.ValueParam("$organization_id", types.TextValue(params.OrganizationID))) + } + return table.NewQueryParameters(opts...) +} + +func scanCoursesResult(ctx context.Context, res result.Result, result *domain.ListCoursesResult) error { + if !res.NextResultSet(ctx) || !res.HasNextRow() { + return nil + } + for res.NextRow() { + var cdb courseDB + if err := res.ScanNamed(cdb.namedValues()...); err != nil { + return fmt.Errorf("scanning row: %w", err) + } + result.Courses = append(result.Courses, mapCourseDB(cdb)) + } + if err := res.Err(); err != nil { + return err + } + if len(result.Courses) > 0 { + result.NextPageToken = result.Courses[len(result.Courses)-1].ID + } + return nil +} + +// --------------------------------------------------------------------------- +// Struct value builders for BulkUpsert +// --------------------------------------------------------------------------- + +func createCourseAsStructValue(params domain.CreateCourseParams) types.Value { + now := time.Now() + st := mapSourceTypeFromDomain(params.SourceType) + + return types.StructValue( + types.StructFieldValue("id", types.TextValue(params.ID)), + types.StructFieldValue("external_id", nullableTextValue(params.ExternalID)), + types.StructFieldValue("name", types.TextValue(params.Name)), + types.StructFieldValue("source_type", types.TextValue(st)), + types.StructFieldValue("source_name", nullableTextValue(params.SourceName)), + types.StructFieldValue("course_thematic", types.TextValue(params.CourseThematic)), + types.StructFieldValue("learning_type", types.TextValue(params.LearningType)), + types.StructFieldValue("organization_id", types.TextValue(params.OrganizationID)), + types.StructFieldValue("origin_link", types.TextValue(params.OriginLink)), + types.StructFieldValue("image_link", types.TextValue(params.ImageLink)), + types.StructFieldValue("description", types.TextValue(params.Description)), + types.StructFieldValue("full_price", types.DoubleValue(params.FullPrice)), + types.StructFieldValue("discount", types.DoubleValue(params.Discount)), + types.StructFieldValue("duration", types.IntervalValueFromDuration(params.Duration)), + types.StructFieldValue("starts_at", types.DatetimeValueFromTime(params.StartsAt)), + types.StructFieldValue("created_at", types.DatetimeValueFromTime(now)), + types.StructFieldValue("updated_at", types.DatetimeValueFromTime(now)), + types.StructFieldValue("deleted_at", types.NullValue(types.Optional(types.TypeDatetime))), + ) +} + +func nullableTextValue(v nullable.Value[string]) types.Value { + if v.Valid() { + return types.OptionalValue(types.TextValue(v.Value())) + } + return types.NullValue(types.Optional(types.TypeText)) +} diff --git a/internal/kurious/service/service.go b/internal/kurious/service/service.go index a0f3f69..b4f436c 100644 --- a/internal/kurious/service/service.go +++ b/internal/kurious/service/service.go @@ -55,7 +55,14 @@ func NewApplication(ctx context.Context, cfg ApplicationConfig, mapper domain.Co organizationrepo = sqliteConnection.Organization() repoCloser = sqliteConnection case RepositoryEngineYDB: - return Application{}, errors.New("ydb is no longer supported") + ydbConn, err := adapters.NewYDBConnection(ctx, cfg.YDB, log.With(slog.String("db", "ydb"))) + if err != nil { + return Application{}, fmt.Errorf("making ydb connection: %w", err) + } + + courseadapter = ydbConn.CourseRepository() + organizationrepo = ydbConn.Organization() + repoCloser = ydbConn default: return Application{}, errors.New("unable to decide which db engine to use") } -- 2.34.1