postgres.NewSchemaReader - Code Metrics - Inspection of "Merge pull request #1153 from Permify/feature/part..." - Permify/permify - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 496048...563bb3 )

by Tolga

created 2024-03-24 14:10 UTC

postgres.NewSchemaReader A

↳ Parent: internal/storage/postgres/schemaReader.go

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	4
rs	10
c	0
b	0
f	0

package postgres

import (
	"context"
	"database/sql"
	"errors"
	"log/slog"

	"github.com/Masterminds/squirrel"
	"github.com/rs/xid"

	"github.com/Permify/permify/internal/schema"
	"github.com/Permify/permify/internal/storage"
	"github.com/Permify/permify/internal/storage/postgres/utils"
	"github.com/Permify/permify/pkg/database"
	db "github.com/Permify/permify/pkg/database/postgres"
	base "github.com/Permify/permify/pkg/pb/base/v1"
)

// SchemaReader - Structure for SchemaReader
type SchemaReader struct {
	database *db.Postgres
	// options
	txOptions sql.TxOptions
}

// NewSchemaReader - Creates a new SchemaReader
func NewSchemaReader(database *db.Postgres) *SchemaReader {
	return &SchemaReader{
		database:  database,
		txOptions: sql.TxOptions{Isolation: sql.LevelReadCommitted, ReadOnly: true},
	}
}

// ReadSchema returns the schema definition for a specific tenant and version as a structured object.
func (r *SchemaReader) ReadSchema(ctx context.Context, tenantID, version string) (sch *base.SchemaDefinition, err error) {
	ctx, span := tracer.Start(ctx, "schema-reader.read-schema")
	defer span.End()

	slog.Debug("reading schema", slog.Any("tenant_id", tenantID), slog.Any("version", version))

	builder := r.database.Builder.Select("name, serialized_definition, version").From(SchemaDefinitionTable).Where(squirrel.Eq{"version": version, "tenant_id": tenantID})

	var query string
	var args []interface{}

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

	slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))

	var rows *sql.Rows
	rows, err = r.database.DB.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
	}
	defer rows.Close()

	var definitions []string
	for rows.Next() {
		sd := storage.SchemaDefinition{}
		err = rows.Scan(&sd.Name, &sd.SerializedDefinition, &sd.Version)
		if err != nil {
			return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
		}
		definitions = append(definitions, sd.Serialized())
	}
	if err = rows.Err(); err != nil {
		return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
	}

	slog.Debug("successfully retrieved", slog.Any("schema definitions", len(definitions)))

	sch, err = schema.NewSchemaFromStringDefinitions(false, definitions...)
	if err != nil {
		return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
	}

	return sch, err
}

// ReadSchemaString returns the schema definition for a specific tenant and version as a string.
func (r *SchemaReader) ReadSchemaString(ctx context.Context, tenantID, version string) (definitions []string, err error) {
	ctx, span := tracer.Start(ctx, "schema-reader.read-schema-string")
	defer span.End()

	slog.Debug("reading schema", slog.Any("tenant_id", tenantID), slog.Any("version", version))

	builder := r.database.Builder.Select("name, serialized_definition, version").From(SchemaDefinitionTable).Where(squirrel.Eq{"version": version, "tenant_id": tenantID})

	var query string
	var args []interface{}

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

	slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))

	var rows *sql.Rows
	rows, err = r.database.DB.QueryContext(ctx, query, args...)
	if err != nil {
		return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
	}
	defer rows.Close()

	for rows.Next() {
		sd := storage.SchemaDefinition{}
		err = rows.Scan(&sd.Name, &sd.SerializedDefinition, &sd.Version)
		if err != nil {
			return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
		}
		definitions = append(definitions, sd.Serialized())
	}
	if err = rows.Err(); err != nil {
		return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
	}

	slog.Debug("successfully retrieved", slog.Any("schema definitions", len(definitions)))

	return definitions, err
}

// ReadEntityDefinition - Reads entity config from the repository.
func (r *SchemaReader) ReadEntityDefinition(ctx context.Context, tenantID, name, version string) (definition *base.EntityDefinition, v string, err error) {
	ctx, span := tracer.Start(ctx, "schema-reader.read-entity-definition")
	defer span.End()

	slog.Debug("reading entity definition", slog.Any("tenant_id", tenantID), slog.Any("version", version))

	builder := r.database.Builder.Select("name, serialized_definition, version").Where(squirrel.Eq{"name": name, "version": version, "tenant_id": tenantID}).From(SchemaDefinitionTable).Limit(1)

	var query string
	var args []interface{}

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

	slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))

	var def storage.SchemaDefinition
	row := r.database.DB.QueryRowContext(ctx, query, args...)
	if err = row.Err(); err != nil {
		return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
	}

	if err = row.Scan(&def.Name, &def.SerializedDefinition, &def.Version); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCHEMA_NOT_FOUND)
		}
		return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
	}

	var sch *base.SchemaDefinition
	sch, err = schema.NewSchemaFromStringDefinitions(false, def.Serialized())
	if err != nil {
		return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
	}

	definition, err = schema.GetEntityByName(sch, name)

	slog.Debug("successfully retrieved", slog.Any("schema definition", definition))

	return definition, def.Version, err
}

// ReadRuleDefinition - Reads rule config from the repository.
func (r *SchemaReader) ReadRuleDefinition(ctx context.Context, tenantID, name, version string) (definition *base.RuleDefinition, v string, err error) {
	ctx, span := tracer.Start(ctx, "schema-reader.read-rule-definition")
	defer span.End()

	slog.Debug("reading rule definition", slog.Any("tenant_id", tenantID), slog.Any("name", name), slog.Any("version", version))

	builder := r.database.Builder.Select("name, serialized_definition, version").Where(squirrel.Eq{"name": name, "version": version, "tenant_id": tenantID}).From(SchemaDefinitionTable).Limit(1)

	var query string
	var args []interface{}

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

	slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))

	var def storage.SchemaDefinition
	row := r.database.DB.QueryRowContext(ctx, query, args...)
	if err = row.Err(); err != nil {
		return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
	}

	if err = row.Scan(&def.Name, &def.SerializedDefinition, &def.Version); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCHEMA_NOT_FOUND)
		}
		return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
	}

	slog.Debug("successfully retrieved rule definition for", slog.Any("name", name))

	var sch *base.SchemaDefinition
	sch, err = schema.NewSchemaFromStringDefinitions(false, def.Serialized())
	if err != nil {
		return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
	}

	definition, err = schema.GetRuleByName(sch, name)

	slog.Debug("successfully created rule definition")

	return definition, def.Version, err
}

// HeadVersion - Finds the latest version of the schema.
func (r *SchemaReader) HeadVersion(ctx context.Context, tenantID string) (version string, err error) {
	ctx, span := tracer.Start(ctx, "schema-reader.head-version")
	defer span.End()

	slog.Debug("finding the latest version fo the schema for", slog.String("tenant_id", tenantID))

	var query string
	var args []interface{}
	query, args, err = r.database.Builder.
		Select("version").From(SchemaDefinitionTable).Where(squirrel.Eq{"tenant_id": tenantID}).OrderBy("version DESC").Limit(1).
		ToSql()
	if err != nil {
		return "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
	}

	slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))

	row := r.database.DB.QueryRowContext(ctx, query, args...)
	err = row.Scan(&version)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCHEMA_NOT_FOUND)
		}
		return "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
	}

	slog.Debug("successfully found the latest schema version", slog.Any("version", version))

	return version, nil
}

// 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 := tracer.Start(ctx, "tenant-reader.list-tenants")
	defer span.End()

	slog.Debug("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_INTERNAL)
		}
		builder = builder.Where(squirrel.GtOrEq{"version": t.(utils.ContinuousToken).Value})
	}

	builder = builder.OrderBy("version").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.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))

	var rows *sql.Rows
	rows, err = r.database.DB.QueryContext(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() {
		schema := &base.SchemaList{}
		err = rows.Scan(&schema.Version)
		if err != nil {
			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
		}
		id, err := xid.FromString(schema.Version)
		if err != nil {
			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
		}
		schema.CreatedAt = id.Time().String()
		lastVersion = schema.Version
		schemas = append(schemas, schema)
	}
	if err = rows.Err(); err != nil {
		return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
	}

	slog.Debug("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
	}
	return schemas, database.NewNoopContinuousToken().Encode(), nil
}


1			package postgres
2
3			import (
4			"context"
5			"database/sql"
6			"errors"
7			"log/slog"
8
9			"github.com/Masterminds/squirrel"
10			"github.com/rs/xid"
11
12			"github.com/Permify/permify/internal/schema"
13			"github.com/Permify/permify/internal/storage"
14			"github.com/Permify/permify/internal/storage/postgres/utils"
15			"github.com/Permify/permify/pkg/database"
16			db "github.com/Permify/permify/pkg/database/postgres"
17			base "github.com/Permify/permify/pkg/pb/base/v1"
18			)
19
20			// SchemaReader - Structure for SchemaReader
21			type SchemaReader struct {
22			database *db.Postgres
23			// options
24			txOptions sql.TxOptions
25			}
26
27			// NewSchemaReader - Creates a new SchemaReader
28			func NewSchemaReader(database db.Postgres) SchemaReader {
29			return &SchemaReader{
30			database: database,
31			txOptions: sql.TxOptions{Isolation: sql.LevelReadCommitted, ReadOnly: true},
32			}
33			}
34
35			// ReadSchema returns the schema definition for a specific tenant and version as a structured object.
36			func (r SchemaReader) ReadSchema(ctx context.Context, tenantID, version string) (sch base.SchemaDefinition, err error) {
37			ctx, span := tracer.Start(ctx, "schema-reader.read-schema")
38			defer span.End()
39
40			slog.Debug("reading schema", slog.Any("tenant_id", tenantID), slog.Any("version", version))
41
42			builder := r.database.Builder.Select("name, serialized_definition, version").From(SchemaDefinitionTable).Where(squirrel.Eq{"version": version, "tenant_id": tenantID})
43
44			var query string
45			var args []interface{}
46
47			query, args, err = builder.ToSql()
48			if err != nil {
49			return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
50			}
51
52			slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))
53
54			var rows *sql.Rows
55			rows, err = r.database.DB.QueryContext(ctx, query, args...)
56			if err != nil {
57			return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
58			}
59			defer rows.Close()
60
61			var definitions []string
62			for rows.Next() {
63			sd := storage.SchemaDefinition{}
64			err = rows.Scan(&sd.Name, &sd.SerializedDefinition, &sd.Version)
65			if err != nil {
66			return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
67			}
68			definitions = append(definitions, sd.Serialized())
69			}
70			if err = rows.Err(); err != nil {
71			return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
72			}
73
74			slog.Debug("successfully retrieved", slog.Any("schema definitions", len(definitions)))
75
76			sch, err = schema.NewSchemaFromStringDefinitions(false, definitions...)
77			if err != nil {
78			return nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
79			}
80
81			return sch, err
82			}
83
84			// ReadSchemaString returns the schema definition for a specific tenant and version as a string.
85			func (r *SchemaReader) ReadSchemaString(ctx context.Context, tenantID, version string) (definitions []string, err error) {
86			ctx, span := tracer.Start(ctx, "schema-reader.read-schema-string")
87			defer span.End()
88
89			slog.Debug("reading schema", slog.Any("tenant_id", tenantID), slog.Any("version", version))
90
91			builder := r.database.Builder.Select("name, serialized_definition, version").From(SchemaDefinitionTable).Where(squirrel.Eq{"version": version, "tenant_id": tenantID})
92
93			var query string
94			var args []interface{}
95
96			query, args, err = builder.ToSql()
97			if err != nil {
98			return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
99			}
100
101			slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))
102
103			var rows *sql.Rows
104			rows, err = r.database.DB.QueryContext(ctx, query, args...)
105			if err != nil {
106			return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
107			}
108			defer rows.Close()
109
110			for rows.Next() {
111			sd := storage.SchemaDefinition{}
112			err = rows.Scan(&sd.Name, &sd.SerializedDefinition, &sd.Version)
113			if err != nil {
114			return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
115			}
116			definitions = append(definitions, sd.Serialized())
117			}
118			if err = rows.Err(); err != nil {
119			return []string{}, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
120			}
121
122			slog.Debug("successfully retrieved", slog.Any("schema definitions", len(definitions)))
123
124			return definitions, err
125			}
126
127			// ReadEntityDefinition - Reads entity config from the repository.
128			func (r SchemaReader) ReadEntityDefinition(ctx context.Context, tenantID, name, version string) (definition base.EntityDefinition, v string, err error) {
129			ctx, span := tracer.Start(ctx, "schema-reader.read-entity-definition")
130			defer span.End()
131
132			slog.Debug("reading entity definition", slog.Any("tenant_id", tenantID), slog.Any("version", version))
133
134			builder := r.database.Builder.Select("name, serialized_definition, version").Where(squirrel.Eq{"name": name, "version": version, "tenant_id": tenantID}).From(SchemaDefinitionTable).Limit(1)
135
136			var query string
137			var args []interface{}
138
139			query, args, err = builder.ToSql()
140			if err != nil {
141			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
142			}
143
144			slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))
145
146			var def storage.SchemaDefinition
147			row := r.database.DB.QueryRowContext(ctx, query, args...)
148			if err = row.Err(); err != nil {
149			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
150			}
151
152			if err = row.Scan(&def.Name, &def.SerializedDefinition, &def.Version); err != nil {
153			if errors.Is(err, sql.ErrNoRows) {
154			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCHEMA_NOT_FOUND)
155			}
156			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
157			}
158
159			var sch *base.SchemaDefinition
160			sch, err = schema.NewSchemaFromStringDefinitions(false, def.Serialized())
161			if err != nil {
162			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
163			}
164
165			definition, err = schema.GetEntityByName(sch, name)
166
167			slog.Debug("successfully retrieved", slog.Any("schema definition", definition))
168
169			return definition, def.Version, err
170			}
171
172			// ReadRuleDefinition - Reads rule config from the repository.
173			func (r SchemaReader) ReadRuleDefinition(ctx context.Context, tenantID, name, version string) (definition base.RuleDefinition, v string, err error) {
174			ctx, span := tracer.Start(ctx, "schema-reader.read-rule-definition")
175			defer span.End()
176
177			slog.Debug("reading rule definition", slog.Any("tenant_id", tenantID), slog.Any("name", name), slog.Any("version", version))
178
179			builder := r.database.Builder.Select("name, serialized_definition, version").Where(squirrel.Eq{"name": name, "version": version, "tenant_id": tenantID}).From(SchemaDefinitionTable).Limit(1)
180
181			var query string
182			var args []interface{}
183
184			query, args, err = builder.ToSql()
185			if err != nil {
186			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
187			}
188
189			slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))
190
191			var def storage.SchemaDefinition
192			row := r.database.DB.QueryRowContext(ctx, query, args...)
193			if err = row.Err(); err != nil {
194			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
195			}
196
197			if err = row.Scan(&def.Name, &def.SerializedDefinition, &def.Version); err != nil {
198			if errors.Is(err, sql.ErrNoRows) {
199			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCHEMA_NOT_FOUND)
200			}
201			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
202			}
203
204			slog.Debug("successfully retrieved rule definition for", slog.Any("name", name))
205
206			var sch *base.SchemaDefinition
207			sch, err = schema.NewSchemaFromStringDefinitions(false, def.Serialized())
208			if err != nil {
209			return nil, "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
210			}
211
212			definition, err = schema.GetRuleByName(sch, name)
213
214			slog.Debug("successfully created rule definition")
215
216			return definition, def.Version, err
217			}
218
219			// HeadVersion - Finds the latest version of the schema.
220			func (r *SchemaReader) HeadVersion(ctx context.Context, tenantID string) (version string, err error) {
221			ctx, span := tracer.Start(ctx, "schema-reader.head-version")
222			defer span.End()
223
224			slog.Debug("finding the latest version fo the schema for", slog.String("tenant_id", tenantID))
225
226			var query string
227			var args []interface{}
228			query, args, err = r.database.Builder.
229			Select("version").From(SchemaDefinitionTable).Where(squirrel.Eq{"tenant_id": tenantID}).OrderBy("version DESC").Limit(1).
230			ToSql()
231			if err != nil {
232			return "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
233			}
234
235			slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))
236
237			row := r.database.DB.QueryRowContext(ctx, query, args...)
238			err = row.Scan(&version)
239			if err != nil {
240			if errors.Is(err, sql.ErrNoRows) {
241			return "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCHEMA_NOT_FOUND)
242			}
243			return "", utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
244			}
245
246			slog.Debug("successfully found the latest schema version", slog.Any("version", version))
247
248			return version, nil
249			}
250
251			// ListSchemas - List all Schemas
252			func (r SchemaReader) ListSchemas(ctx context.Context, tenantID string, pagination database.Pagination) (schemas []base.SchemaList, ct database.EncodedContinuousToken, err error) {
253			ctx, span := tracer.Start(ctx, "tenant-reader.list-tenants")
254			defer span.End()
255
256			slog.Debug("listing schemas with pagination", slog.Any("pagination", pagination))
257
258			builder := r.database.Builder.Select("DISTINCT version").From(SchemaDefinitionTable).Where(squirrel.Eq{"tenant_id": tenantID})
259			if pagination.Token() != "" {
260			var t database.ContinuousToken
261			t, err = utils.EncodedContinuousToken{Value: pagination.Token()}.Decode()
262			if err != nil {
263			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
264			}
265			builder = builder.Where(squirrel.GtOrEq{"version": t.(utils.ContinuousToken).Value})
266			}
267
268			builder = builder.OrderBy("version").Limit(uint64(pagination.PageSize() + 1))
269
270			var query string
271			var args []interface{}
272
273			query, args, err = builder.ToSql()
274			if err != nil {
275			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER)
276			}
277
278			slog.Debug("executing sql query", slog.Any("query", query), slog.Any("arguments", args))
279
280			var rows *sql.Rows
281			rows, err = r.database.DB.QueryContext(ctx, query, args...)
282			if err != nil {
283			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION)
284			}
285			defer rows.Close()
286
287			var lastVersion string
288			schemas = make([]*base.SchemaList, 0, pagination.PageSize()+1)
289			for rows.Next() {
290			schema := &base.SchemaList{}
291			err = rows.Scan(&schema.Version)
292			if err != nil {
293			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
294			}
295			id, err := xid.FromString(schema.Version)
296			if err != nil {
297			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SCAN)
298			}
299			schema.CreatedAt = id.Time().String()
300			lastVersion = schema.Version
301			schemas = append(schemas, schema)
302			}
303			if err = rows.Err(); err != nil {
304			return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL)
305			}
306
307			slog.Debug("successfully listed schemas", slog.Any("number_of_schemas", len(schemas)))
308
309			if len(schemas) > int(pagination.PageSize()) {
310			return schemas[:pagination.PageSize()], utils.NewContinuousToken(lastVersion).Encode(), nil
311			}
312			return schemas, database.NewNoopContinuousToken().Encode(), nil
313			}
314

Permify / permify

Push — master ( 496048...563bb3 )

postgres.NewSchemaReader A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like