Method ListSchemas

internal/storage/postgres/schema_reader.go:221–281

ListSchemas - List all Schemas

(ctx context.Context, tenantID string, pagination database.Pagination)

// ListSchemas - List all Schemas
func (r *SchemaReader) ListSchemas(ctx context.Context, tenantID string, pagination database.Pagination) (schemas []*base.SchemaList, ct database.EncodedContinuousToken, err error) {
	ctx, span := internal.Tracer.Start(ctx, "tenant-reader.list-tenants")
	defer span.End()

	slog.DebugContext(ctx, "listing schemas with pagination", slog.Any("pagination", pagination))

	builder := r.database.Builder.Select("DISTINCT version").From(SchemaDefinitionTable).Where(squirrel.Eq{"tenant_id": tenantID})
	if pagination.Token() != "" {
		var t database.ContinuousToken
		t, err = utils.EncodedContinuousToken{Value: pagination.Token()}.Decode()
		if err != nil {
			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INVALID_CONTINUOUS_TOKEN)
		}
		builder = builder.Where(squirrel.LtOrEq{"version": t.(utils.ContinuousToken).Value})
	}

	builder = builder.OrderBy("version DESC").Limit(uint64(pagination.PageSize() + 1))

	var query string
	var args []interface{}

	query, args, err = builder.ToSql()
	if err != nil {
		return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
	}

	slog.DebugContext(ctx, "executing sql query", slog.Any("query", query), slog.Any("arguments", args))
	var rows pgx.Rows
	rows, err = r.database.ReadPool.Query(ctx, query, args...)
	if err != nil {
		return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
	}
	defer rows.Close()

	var lastVersion string
	schemas = make([]*base.SchemaList, 0, pagination.PageSize()+1)
	for rows.Next() {
		sch := &base.SchemaList{}
		err = rows.Scan(&sch.Version)
		if err != nil {
			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
		}
		id, err := xid.FromString(sch.Version)
		if err != nil {
			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
		}
		sch.CreatedAt = id.Time().String()
		lastVersion = sch.Version
		schemas = append(schemas, sch)
	}
	if err = rows.Err(); err != nil {
		return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
	}

	slog.DebugContext(ctx, "successfully listed schemas", slog.Any("number_of_schemas", len(schemas)))

	if len(schemas) > int(pagination.PageSize()) {
		return schemas[:pagination.PageSize()], utils.NewContinuousToken(lastVersion).Encode(), nil