mirror of
https://github.com/HeyPuter/puter.git
synced 2026-08-24 15:07:17 +00:00
feat(share): query and maintain active shares in ShareStore
This commit is contained in:
Juan Castro
@@ -27,6 +27,7 @@ import { AuthService } from './auth/AuthService';
|
||||
import { OIDCService } from './auth/OIDCService';
|
||||
import { TokenService } from './auth/TokenService';
|
||||
import { BroadcastService } from './broadcast/BroadcastService';
|
||||
import { CacheReplicationService } from './cache/CacheReplicationService';
|
||||
import { FSService } from './fs/FSService';
|
||||
import { ServerHealthService } from './health/ServerHealthService';
|
||||
import { PuterHomepageService } from './homepage/PuterHomepageService';
|
||||
@@ -62,6 +63,7 @@ declare module './types' {
|
||||
socket: SocketService;
|
||||
notification: NotificationService;
|
||||
broadcast: BroadcastService;
|
||||
cacheReplication: CacheReplicationService;
|
||||
oidc: OIDCService;
|
||||
appIcon: AppIconService;
|
||||
defaultUser: DefaultUserService;
|
||||
@@ -100,6 +102,8 @@ export const puterServices = {
|
||||
socket: SocketService,
|
||||
notification: NotificationService,
|
||||
broadcast: BroadcastService,
|
||||
// Independent — only needs the event client and redis.
|
||||
cacheReplication: CacheReplicationService,
|
||||
oidc: OIDCService,
|
||||
appIcon: AppIconService,
|
||||
defaultUser: DefaultUserService,
|
||||
|
||||
@@ -18,17 +18,26 @@
|
||||
*/
|
||||
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
import { encodeCursor, decodeCursor } from '../../util/pagination';
|
||||
import { PuterStore } from '../types';
|
||||
|
||||
/** Default page size for `listByHolder`. */
|
||||
const DEFAULT_HOLDER_PAGE_SIZE = 50;
|
||||
const MAX_HOLDER_PAGE_SIZE = 200;
|
||||
|
||||
/**
|
||||
* CRUD over the `share` table.
|
||||
*
|
||||
* Columns: id, uid (unique), issuer_user_id, recipient_email, data (JSON),
|
||||
* created_at.
|
||||
* Columns: id, uid (unique), issuer_user_id, recipient_email, holder_user_id,
|
||||
* fsentry_id, mode, data (JSON), created_at, applied_at.
|
||||
*
|
||||
* Shares are pending permission grants sent to an email address. Once the
|
||||
* recipient applies the share, the permissions are granted and the row is
|
||||
* deleted.
|
||||
* The table carries two related things. A row with a `holder_user_id` is an
|
||||
* **active share** — the index that makes shares listable and ties them to an
|
||||
* fsentry so they die with the file. A row without one is a **pending invite**
|
||||
* to an email that has no account yet; claiming it fills in the holder rather
|
||||
* than deleting the row, so the share stays queryable afterwards.
|
||||
*
|
||||
* Permissions remain the source of truth for access. This is the index.
|
||||
*/
|
||||
export class ShareStore extends PuterStore {
|
||||
// -- Reads --------------------------------------------------------
|
||||
@@ -57,6 +66,58 @@ export class ShareStore extends PuterStore {
|
||||
return rows.map((r) => this.#normalizeRow(r));
|
||||
}
|
||||
|
||||
/**
|
||||
* Active shares held by a user, keyset-paginated. `id` is the tiebreaker,
|
||||
* so a row added mid-iteration can't shift earlier pages.
|
||||
*
|
||||
* Returns rows only; the caller hydrates fsentries (batched) and drops any
|
||||
* whose entry it can't resolve. Paths are deliberately not stored — a move
|
||||
* or rename would strand them.
|
||||
*/
|
||||
async listByHolder(holderUserId, { limit, cursor } = {}) {
|
||||
const size = Math.min(
|
||||
Math.max(1, Math.floor(Number(limit) || DEFAULT_HOLDER_PAGE_SIZE)),
|
||||
MAX_HOLDER_PAGE_SIZE,
|
||||
);
|
||||
const decoded = decodeCursor(cursor, 'share cursor');
|
||||
const afterId = Number(decoded?.id ?? 0) || 0;
|
||||
|
||||
// One extra row tells us whether another page exists.
|
||||
const rows = await this.clients.db.read(
|
||||
'SELECT * FROM `share` WHERE `holder_user_id` = ? AND `id` > ? ' +
|
||||
'ORDER BY `id` LIMIT ?',
|
||||
[holderUserId, afterId, size + 1],
|
||||
);
|
||||
|
||||
const hasMore = rows.length > size;
|
||||
const items = (hasMore ? rows.slice(0, size) : rows).map((r) =>
|
||||
this.#normalizeRow(r),
|
||||
);
|
||||
const last = items[items.length - 1];
|
||||
return {
|
||||
items,
|
||||
cursor: hasMore && last ? encodeCursor({ id: last.id }) : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
/** Everyone with an active share on one node, whoever issued it. */
|
||||
async listByFsentry(fsentryId) {
|
||||
const rows = await this.clients.db.read(
|
||||
'SELECT * FROM `share` WHERE `fsentry_id` = ? AND ' +
|
||||
'`holder_user_id` IS NOT NULL ORDER BY `id`',
|
||||
[fsentryId],
|
||||
);
|
||||
return rows.map((r) => this.#normalizeRow(r));
|
||||
}
|
||||
|
||||
async countByHolder(holderUserId) {
|
||||
const rows = await this.clients.db.read(
|
||||
'SELECT COUNT(*) AS `count` FROM `share` WHERE `holder_user_id` = ?',
|
||||
[holderUserId],
|
||||
);
|
||||
return Number(rows[0]?.count ?? 0);
|
||||
}
|
||||
|
||||
// -- Writes -------------------------------------------------------
|
||||
|
||||
async create({ issuerUserId, recipientEmail, data }) {
|
||||
@@ -75,6 +136,94 @@ export class ShareStore extends PuterStore {
|
||||
return this.getByUid(uid);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record an active share, or move an existing one to a new mode. Keyed on
|
||||
* (holder, fsentry, issuer) to match the table's unique index — two people
|
||||
* with manage rights each keep their own row rather than overwriting.
|
||||
*/
|
||||
async upsertActive({
|
||||
issuerUserId,
|
||||
holderUserId,
|
||||
fsentryId,
|
||||
mode,
|
||||
recipientEmail = null,
|
||||
}) {
|
||||
if (!issuerUserId || !holderUserId || !fsentryId || !mode) {
|
||||
throw new Error(
|
||||
'upsertActive: issuerUserId, holderUserId, fsentryId and mode are required',
|
||||
);
|
||||
}
|
||||
|
||||
const existing = await this.clients.db.read(
|
||||
'SELECT `uid` FROM `share` WHERE `holder_user_id` = ? AND ' +
|
||||
'`fsentry_id` = ? AND `issuer_user_id` = ? LIMIT 1',
|
||||
[holderUserId, fsentryId, issuerUserId],
|
||||
);
|
||||
if (existing[0]?.uid) {
|
||||
await this.clients.db.write(
|
||||
'UPDATE `share` SET `mode` = ? WHERE `uid` = ?',
|
||||
[mode, existing[0].uid],
|
||||
);
|
||||
return this.getByUid(existing[0].uid);
|
||||
}
|
||||
|
||||
const uid = uuidv4();
|
||||
await this.clients.db.write(
|
||||
'INSERT INTO `share` (`uid`, `issuer_user_id`, `recipient_email`, ' +
|
||||
'`holder_user_id`, `fsentry_id`, `mode`, `applied_at`) ' +
|
||||
'VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)',
|
||||
[
|
||||
uid,
|
||||
issuerUserId,
|
||||
recipientEmail ?? '',
|
||||
holderUserId,
|
||||
fsentryId,
|
||||
mode,
|
||||
],
|
||||
);
|
||||
return this.getByUid(uid);
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop one active share. Omit `issuerUserId` to clear every issuer's share
|
||||
* of that node with that holder — what an owner revoking access wants.
|
||||
*/
|
||||
async deleteActive({ holderUserId, fsentryId, issuerUserId = null }) {
|
||||
const scoped = issuerUserId !== null && issuerUserId !== undefined;
|
||||
const result = await this.clients.db.write(
|
||||
'DELETE FROM `share` WHERE `holder_user_id` = ? AND `fsentry_id` = ?' +
|
||||
(scoped ? ' AND `issuer_user_id` = ?' : ''),
|
||||
scoped
|
||||
? [holderUserId, fsentryId, issuerUserId]
|
||||
: [holderUserId, fsentryId],
|
||||
);
|
||||
return (result?.affectedRows ?? result?.changes ?? 0) > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Claim a pending invite for the user who signed up. Updates rather than
|
||||
* deletes, so the share survives as an index row.
|
||||
*/
|
||||
async applyPending({ uid, holderUserId, fsentryId = null, mode = null }) {
|
||||
if (!uid || !holderUserId) {
|
||||
throw new Error('applyPending: uid and holderUserId are required');
|
||||
}
|
||||
const result = await this.clients.db.write(
|
||||
'UPDATE `share` SET `holder_user_id` = ?, `applied_at` = CURRENT_TIMESTAMP' +
|
||||
(fsentryId === null ? '' : ', `fsentry_id` = ?') +
|
||||
(mode === null ? '' : ', `mode` = ?') +
|
||||
' WHERE `uid` = ? AND `holder_user_id` IS NULL',
|
||||
[
|
||||
holderUserId,
|
||||
...(fsentryId === null ? [] : [fsentryId]),
|
||||
...(mode === null ? [] : [mode]),
|
||||
uid,
|
||||
],
|
||||
);
|
||||
if ((result?.affectedRows ?? result?.changes ?? 0) === 0) return null;
|
||||
return this.getByUid(uid);
|
||||
}
|
||||
|
||||
async deleteByUid(uid) {
|
||||
const result = await this.clients.db.write(
|
||||
'DELETE FROM `share` WHERE `uid` = ?',
|
||||
|
||||
@@ -233,4 +233,239 @@ describe('ShareStore', () => {
|
||||
false,
|
||||
);
|
||||
});
|
||||
|
||||
// -- active shares (the index) -------------------------------------
|
||||
|
||||
describe('active shares', () => {
|
||||
let holder;
|
||||
|
||||
const makeEntry = async (owner) => {
|
||||
const uuid = uuidv4();
|
||||
await server.clients.db.write(
|
||||
'INSERT INTO `fsentries` (`uuid`, `name`, `path`, `user_id`, `is_dir`, `modified`) VALUES (?, ?, ?, ?, 0, ?)',
|
||||
[
|
||||
uuid,
|
||||
`f-${uuid.slice(0, 8)}`,
|
||||
`/x/${uuid}`,
|
||||
owner.id,
|
||||
Math.floor(Date.now() / 1000),
|
||||
],
|
||||
);
|
||||
const rows = await server.clients.db.read(
|
||||
'SELECT `id` FROM `fsentries` WHERE `uuid` = ?',
|
||||
[uuid],
|
||||
);
|
||||
return { id: Number(rows[0].id), uuid };
|
||||
};
|
||||
|
||||
beforeAll(async () => {
|
||||
holder = await makeUser();
|
||||
});
|
||||
|
||||
it('records an active share and lists it for the holder', async () => {
|
||||
const entry = await makeEntry(issuer);
|
||||
const created = await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
|
||||
expect(created.holder_user_id).toBe(holder.id);
|
||||
expect(created.fsentry_id).toBe(entry.id);
|
||||
expect(created.mode).toBe('read');
|
||||
expect(created.applied_at).toBeTruthy();
|
||||
|
||||
const page = await store.listByHolder(holder.id);
|
||||
expect(page.items.map((r) => r.uid)).toContain(created.uid);
|
||||
});
|
||||
|
||||
it('moves an existing share to a new mode instead of duplicating it', async () => {
|
||||
const entry = await makeEntry(issuer);
|
||||
const first = await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
const second = await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'write',
|
||||
});
|
||||
|
||||
expect(second.uid).toBe(first.uid);
|
||||
expect(second.mode).toBe('write');
|
||||
expect(await store.listByFsentry(entry.id)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('keeps a separate row per issuer on the same node', async () => {
|
||||
const entry = await makeEntry(issuer);
|
||||
await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
await store.upsertActive({
|
||||
issuerUserId: otherIssuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'write',
|
||||
});
|
||||
|
||||
const rows = await store.listByFsentry(entry.id);
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(rows.map((r) => r.issuer_user_id).sort()).toEqual(
|
||||
[issuer.id, otherIssuer.id].sort(),
|
||||
);
|
||||
});
|
||||
|
||||
it('deletes one issuer share, or every issuer share for the holder', async () => {
|
||||
const entry = await makeEntry(issuer);
|
||||
await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
await store.upsertActive({
|
||||
issuerUserId: otherIssuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
|
||||
expect(
|
||||
await store.deleteActive({
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
issuerUserId: issuer.id,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(await store.listByFsentry(entry.id)).toHaveLength(1);
|
||||
|
||||
expect(
|
||||
await store.deleteActive({
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(await store.listByFsentry(entry.id)).toEqual([]);
|
||||
});
|
||||
|
||||
it('retires the share when the file is deleted', async () => {
|
||||
const entry = await makeEntry(issuer);
|
||||
const created = await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
|
||||
await server.clients.db.write(
|
||||
'DELETE FROM `fsentries` WHERE `id` = ?',
|
||||
[entry.id],
|
||||
);
|
||||
|
||||
// The cascade is what stops a deleted file lingering in the
|
||||
// recipient's listing forever.
|
||||
expect(await store.getByUid(created.uid)).toBeNull();
|
||||
});
|
||||
|
||||
it('paginates by keyset and stops without a trailing cursor', async () => {
|
||||
const pageHolder = await makeUser();
|
||||
const uids = [];
|
||||
for (let i = 0; i < 5; i++) {
|
||||
const entry = await makeEntry(issuer);
|
||||
const row = await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: pageHolder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
uids.push(row.uid);
|
||||
}
|
||||
|
||||
const seen = [];
|
||||
let cursor;
|
||||
for (let guard = 0; guard < 10; guard++) {
|
||||
const page = await store.listByHolder(pageHolder.id, {
|
||||
limit: 2,
|
||||
cursor,
|
||||
});
|
||||
seen.push(...page.items.map((r) => r.uid));
|
||||
cursor = page.cursor;
|
||||
if (!cursor) break;
|
||||
}
|
||||
|
||||
expect(seen).toEqual(uids);
|
||||
expect(cursor).toBeUndefined();
|
||||
expect(await store.countByHolder(pageHolder.id)).toBe(5);
|
||||
});
|
||||
|
||||
it('never returns another holder rows', async () => {
|
||||
const stranger = await makeUser();
|
||||
const entry = await makeEntry(issuer);
|
||||
await store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
|
||||
const page = await store.listByHolder(stranger.id);
|
||||
expect(page.items).toEqual([]);
|
||||
});
|
||||
|
||||
it('claims a pending invite without dropping the row', async () => {
|
||||
const entry = await makeEntry(issuer);
|
||||
const pending = await store.create({
|
||||
issuerUserId: issuer.id,
|
||||
recipientEmail: `pending-${uuidv4()}@test.local`,
|
||||
data: {},
|
||||
});
|
||||
|
||||
const applied = await store.applyPending({
|
||||
uid: pending.uid,
|
||||
holderUserId: holder.id,
|
||||
fsentryId: entry.id,
|
||||
mode: 'read',
|
||||
});
|
||||
|
||||
expect(applied.holder_user_id).toBe(holder.id);
|
||||
expect(applied.mode).toBe('read');
|
||||
expect(applied.applied_at).toBeTruthy();
|
||||
// Second claim finds nothing left to claim.
|
||||
expect(
|
||||
await store.applyPending({
|
||||
uid: pending.uid,
|
||||
holderUserId: holder.id,
|
||||
}),
|
||||
).toBeNull();
|
||||
});
|
||||
|
||||
it('excludes pending invites from a holder listing', async () => {
|
||||
const freshHolder = await makeUser();
|
||||
await store.create({
|
||||
issuerUserId: issuer.id,
|
||||
recipientEmail: `unclaimed-${uuidv4()}@test.local`,
|
||||
data: {},
|
||||
});
|
||||
|
||||
const page = await store.listByHolder(freshHolder.id);
|
||||
expect(page.items).toEqual([]);
|
||||
});
|
||||
|
||||
it('rejects an incomplete active share', async () => {
|
||||
await expect(
|
||||
store.upsertActive({
|
||||
issuerUserId: issuer.id,
|
||||
holderUserId: holder.id,
|
||||
mode: 'read',
|
||||
}),
|
||||
).rejects.toThrow('are required');
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user