Skip to content

Commit 9a43947

Browse files
EspressoTrip-v2EspressoTrip-v2
committed
Merge branch 'main' into feature/sdk-reporting
2 parents 80f8db9 + 704553e commit 9a43947

File tree

4 files changed

+93
-43
lines changed

4 files changed

+93
-43
lines changed

‎.changeset/stupid-lizards-vanish.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@powersync/service-sync-rules': patch
3+
---
4+
5+
Sync streams: Fix `auth.parameter()` to use top-level parameters instead of the nested `parameters` object that the legacy `token_parameters` table uses.

‎packages/sync-rules/src/streams/functions.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ export const STREAM_FUNCTIONS: Record<string, Record<string, SqlParameterFunctio
3434
return v.rawTokenPayload;
3535
},
3636
extractJsonParsed: function (v: ParameterValueSet) {
37-
return v.tokenParameters;
37+
return v.parsedTokenPayload;
3838
},
3939
sourceDescription: 'JWT payload as JSON',
4040
sourceDocumentation: 'JWT payload as a JSON string. This is always validated against trusted keys',

‎packages/sync-rules/src/types.ts‎

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -99,13 +99,15 @@ export interface ParameterValueSet {
9999
* JSON string of raw request parameters.
100100
*/
101101
rawTokenPayload: string;
102-
tokenParameters: SqliteJsonRow;
102+
parsedTokenPayload: SqliteJsonRow;
103+
legacyTokenParameters: SqliteJsonRow;
103104

104105
userId: string;
105106
}
106107

107108
export class RequestParameters implements ParameterValueSet {
108-
tokenParameters: SqliteJsonRow;
109+
parsedTokenPayload: SqliteJsonRow;
110+
legacyTokenParameters: SqliteJsonRow;
109111
userParameters: SqliteJsonRow;
110112

111113
/**
@@ -128,7 +130,8 @@ export class RequestParameters implements ParameterValueSet {
128130

129131
constructor(tokenPayload: RequestJwtPayload | RequestParameters, clientParameters?: Record<string, any>) {
130132
if (tokenPayload instanceof RequestParameters) {
131-
this.tokenParameters = tokenPayload.tokenParameters;
133+
this.parsedTokenPayload = tokenPayload.parsedTokenPayload;
134+
this.legacyTokenParameters = tokenPayload.legacyTokenParameters;
132135
this.userParameters = tokenPayload.userParameters;
133136
this.rawUserParameters = tokenPayload.rawUserParameters;
134137
this.rawTokenPayload = tokenPayload.rawTokenPayload;
@@ -149,7 +152,11 @@ export class RequestParameters implements ParameterValueSet {
149152

150153
// Client and token parameters don't contain DateTime values or other custom types, so we don't need to consider
151154
// compatibility.
152-
this.tokenParameters = toSyncRulesParameters(tokenParameters, CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY);
155+
this.parsedTokenPayload = tokenPayload;
156+
this.legacyTokenParameters = toSyncRulesParameters(
157+
tokenParameters,
158+
CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY
159+
);
153160
this.userId = tokenPayload.sub;
154161
this.rawTokenPayload = JSONBig.stringify(tokenPayload);
155162

@@ -161,7 +168,7 @@ export class RequestParameters implements ParameterValueSet {
161168

162169
lookup(table: string, column: string): SqliteJsonValue {
163170
if (table == 'token_parameters') {
164-
return this.tokenParameters[column];
171+
return this.legacyTokenParameters[column];
165172
} else if (table == 'user_parameters') {
166173
return this.userParameters[column];
167174
} else if (table == 'subscription_parameters' && this.streamParameters != null) {

‎packages/sync-rules/test/src/streams.test.ts‎

Lines changed: 75 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,11 @@ import {
66
CompatibilityEdition,
77
DEFAULT_TAG,
88
GetBucketParameterQuerierResult,
9+
GetQuerierOptions,
910
mergeBucketParameterQueriers,
1011
ParameterLookup,
1112
QuerierError,
13+
RequestParameters,
1214
SourceTableInterface,
1315
SqliteJsonRow,
1416
SqliteRow,
@@ -63,6 +65,36 @@ describe('streams', () => {
6365
]);
6466
});
6567

68+
test('legacy token parameter', async () => {
69+
const desc = parseStream(`SELECT * FROM issues WHERE owner_id = auth.parameter('$.parameters.test')`);
70+
71+
const queriers: BucketParameterQuerier[] = [];
72+
const errors: QuerierError[] = [];
73+
const pending = { queriers, errors };
74+
desc.pushBucketParameterQueriers(
75+
pending,
76+
normalizeQuerierOptions(
77+
{ test: 'foo' },
78+
{},
79+
{ stream: [{ opaque_id: 0, parameters: null }] },
80+
bucketIdTransformer
81+
)
82+
);
83+
84+
expect(mergeBucketParameterQueriers(queriers).staticBuckets).toEqual([
85+
{
86+
bucket: '1#stream|0["foo"]',
87+
definition: 'stream',
88+
inclusion_reasons: [
89+
{
90+
subscription: 0
91+
}
92+
],
93+
priority: 3
94+
}
95+
]);
96+
});
97+
6698
describe('or', () => {
6799
test('parameter match or request condition', async () => {
68100
const desc = parseStream("SELECT * FROM issues WHERE owner_id = auth.user_id() OR auth.parameter('is_admin')");
@@ -74,17 +106,17 @@ describe('streams', () => {
74106

75107
expect(
76108
await queryBucketIds(desc, {
77-
token_parameters: {
78-
user_id: 'u1',
109+
token: {
110+
sub: 'u1',
79111
is_admin: false
80112
}
81113
})
82114
).toStrictEqual(['1#stream|0["u1"]']);
83115

84116
expect(
85117
await queryBucketIds(desc, {
86-
token_parameters: {
87-
user_id: 'u1',
118+
token: {
119+
sub: 'u1',
88120
is_admin: true
89121
}
90122
})
@@ -103,8 +135,8 @@ describe('streams', () => {
103135

104136
expect(
105137
await queryBucketIds(desc, {
106-
token_parameters: {
107-
user_id: 'u1'
138+
token: {
139+
sub: 'u1'
108140
}
109141
})
110142
).toStrictEqual(['1#stream|0["u1"]', '1#stream|1[]']);
@@ -121,14 +153,14 @@ describe('streams', () => {
121153

122154
expect(
123155
await queryBucketIds(desc, {
124-
token_parameters: {
156+
token: {
125157
is_admin: false
126158
}
127159
})
128160
).toStrictEqual(['1#stream|0[]']);
129161
expect(
130162
await queryBucketIds(desc, {
131-
token_parameters: {
163+
token: {
132164
is_admin: true
133165
}
134166
})
@@ -162,15 +194,15 @@ describe('streams', () => {
162194
expect(evaluateBucketIds(desc, COMMENTS, { id: 'foo', content: 'whatever]' })).toStrictEqual(['1#stream|0[]']);
163195
expect(
164196
await queryBucketIds(desc, {
165-
token_parameters: {
197+
token: {
166198
a: false,
167199
b: false
168200
}
169201
})
170202
).toStrictEqual([]);
171203
expect(
172204
await queryBucketIds(desc, {
173-
token_parameters: {
205+
token: {
174206
a: true,
175207
b: false
176208
}
@@ -204,12 +236,13 @@ describe('streams', () => {
204236

205237
return [{ result: 'i1' }];
206238
}
207-
expect(
208-
await queryBucketIds(desc, { token_parameters: { user_id: 'u1', is_admin: false }, getParameterSets })
209-
).toStrictEqual(['1#stream|0["i1"]']);
210-
expect(
211-
await queryBucketIds(desc, { token_parameters: { user_id: 'u1', is_admin: true }, getParameterSets })
212-
).toStrictEqual(['1#stream|1[]', '1#stream|0["i1"]']);
239+
expect(await queryBucketIds(desc, { token: { sub: 'u1', is_admin: false }, getParameterSets })).toStrictEqual([
240+
'1#stream|0["i1"]'
241+
]);
242+
expect(await queryBucketIds(desc, { token: { sub: 'u1', is_admin: true }, getParameterSets })).toStrictEqual([
243+
'1#stream|1[]',
244+
'1#stream|0["i1"]'
245+
]);
213246
});
214247
});
215248

@@ -236,7 +269,7 @@ describe('streams', () => {
236269

237270
expect(
238271
await queryBucketIds(desc, {
239-
token_parameters: { user_id: 'user1' },
272+
token: { sub: 'user1' },
240273
getParameterSets(lookups) {
241274
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['user1'])]);
242275

@@ -267,7 +300,7 @@ describe('streams', () => {
267300
// Should return bucket id for admin users
268301
expect(
269302
await queryBucketIds(desc, {
270-
token_parameters: { user_id: 'u' },
303+
token: { sub: 'u' },
271304
getParameterSets: (lookups: ParameterLookup[]) => {
272305
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['u'])]);
273306
return [{ result: 'u' }];
@@ -278,7 +311,7 @@ describe('streams', () => {
278311
// And not for others
279312
expect(
280313
await queryBucketIds(desc, {
281-
token_parameters: { user_id: 'u2' },
314+
token: { sub: 'u2' },
282315
getParameterSets: (lookups: ParameterLookup[]) => {
283316
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['u2'])]);
284317
return [];
@@ -331,7 +364,7 @@ describe('streams', () => {
331364

332365
expect(
333366
await queryBucketIds(desc, {
334-
token_parameters: { user_id: 'a' },
367+
token: { sub: 'a' },
335368
getParameterSets
336369
})
337370
).toStrictEqual(['1#stream|1["b"]']);
@@ -343,7 +376,7 @@ describe('streams', () => {
343376
expect(evaluateBucketIds(desc, COMMENTS, { id: 'a', issue_id: 'i' })).toStrictEqual(['1#stream|0["i"]']);
344377
expect(
345378
await queryBucketIds(desc, {
346-
token_parameters: { user_id: 'a' },
379+
token: { sub: 'a' },
347380
parameters: { issue_id: ['i1', 'i2'] }
348381
})
349382
).toStrictEqual(['1#stream|0["i1"]', '1#stream|0["i2"]']);
@@ -359,7 +392,7 @@ describe('streams', () => {
359392
]);
360393
expect(
361394
await queryBucketIds(desc, {
362-
token_parameters: { user_id: 'a' },
395+
token: { sub: 'a' },
363396
parameters: { labels: ['l1', 'l2'] },
364397
getParameterSets(lookups) {
365398
expect(lookups).toHaveLength(1);
@@ -401,7 +434,7 @@ describe('streams', () => {
401434

402435
expect(
403436
await queryBucketIds(desc, {
404-
token_parameters: { user_id: 'user1' },
437+
token: { sub: 'user1' },
405438
getParameterSets(lookups) {
406439
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['user1'])]);
407440

@@ -504,7 +537,7 @@ describe('streams', () => {
504537

505538
expect(
506539
await queryBucketIds(desc, {
507-
token_parameters: { user_id: 'user1' },
540+
token: { sub: 'user1' },
508541
getParameterSets(lookups) {
509542
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['user1'])]);
510543

@@ -559,7 +592,7 @@ describe('streams', () => {
559592

560593
expect(
561594
await queryBucketIds(desc, {
562-
token_parameters: { user_id: 'user1' },
595+
token: { sub: 'user1' },
563596
getParameterSets(lookups) {
564597
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['user1'])]);
565598

@@ -569,7 +602,7 @@ describe('streams', () => {
569602
).toStrictEqual(['1#stream|0["issue_id"]']);
570603
expect(
571604
await queryBucketIds(desc, {
572-
token_parameters: { user_id: 'user1', is_admin: true },
605+
token: { sub: 'user1', is_admin: true },
573606
getParameterSets(lookups) {
574607
expect(lookups).toStrictEqual([ParameterLookup.normalized('stream', '0', ['user1'])]);
575608

@@ -654,7 +687,7 @@ function evaluateBucketIds(stream: SyncStream, sourceTable: SourceTableInterface
654687
async function createQueriers(
655688
stream: SyncStream,
656689
options?: {
657-
token_parameters?: Record<string, any>;
690+
token?: Record<string, any>;
658691
parameters?: Record<string, any>;
659692
getParameterSets?: (lookups: ParameterLookup[]) => SqliteJsonRow[];
660693
}
@@ -663,23 +696,28 @@ async function createQueriers(
663696
const errors: QuerierError[] = [];
664697
const pending = { queriers, errors };
665698

666-
stream.pushBucketParameterQueriers(
667-
pending,
668-
normalizeQuerierOptions(
669-
options?.token_parameters ?? {},
670-
{},
671-
{ stream: [{ opaque_id: 0, parameters: options?.parameters ?? null }] },
672-
bucketIdTransformer
673-
)
674-
);
699+
const querierOptions: GetQuerierOptions = {
700+
hasDefaultStreams: true,
701+
globalParameters: new RequestParameters(
702+
{
703+
sub: 'test-user',
704+
...options?.token
705+
},
706+
{}
707+
),
708+
streams: { stream: [{ opaque_id: 0, parameters: options?.parameters ?? null }] },
709+
bucketIdTransformer
710+
};
711+
712+
stream.pushBucketParameterQueriers(pending, querierOptions);
675713

676714
return { querier: mergeBucketParameterQueriers(queriers), errors };
677715
}
678716

679717
async function queryBucketIds(
680718
stream: SyncStream,
681719
options?: {
682-
token_parameters?: Record<string, any>;
720+
token?: Record<string, any>;
683721
parameters?: Record<string, any>;
684722
getParameterSets?: (lookups: ParameterLookup[]) => SqliteJsonRow[];
685723
}

0 commit comments

Comments
 (0)