diff --git a/fluxer_api/src/api/admin/services/AdminGuildService.ts b/fluxer_api/src/api/admin/services/AdminGuildService.ts index e58e463bd..8721a8d88 100644 --- a/fluxer_api/src/api/admin/services/AdminGuildService.ts +++ b/fluxer_api/src/api/admin/services/AdminGuildService.ts @@ -57,6 +57,7 @@ export class AdminGuildService { entityAssetService: deps.entityAssetService, auditService: deps.auditService, updatePropagator: this.updatePropagator, + guildService: deps.guildService, }); this.vanityService = new AdminGuildVanityService({ guildRepository: deps.guildRepository, diff --git a/fluxer_api/src/api/admin/services/guild/AdminGuildUpdateService.ts b/fluxer_api/src/api/admin/services/guild/AdminGuildUpdateService.ts index 17f55ab1c..2b2ccc420 100644 --- a/fluxer_api/src/api/admin/services/guild/AdminGuildUpdateService.ts +++ b/fluxer_api/src/api/admin/services/guild/AdminGuildUpdateService.ts @@ -6,6 +6,9 @@ import type {AdminGuildUpdatePropagator} from '@app/api/admin/services/guild/Adm import {createGuildID, createUserID, type GuildID, type UserID} from '@app/api/BrandedTypes'; import type {GuildRow} from '@app/api/database/types/GuildTypes'; import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; +import {staffEditTouchesInvitePause} from '@app/api/guild/services/GuildInvitePauseApplier'; +import type {GuildService} from '@app/api/guild/services/GuildService'; +import {emitGuildFeaturesChanged} from '@app/api/infrastructure/activity/GuildEvents'; import type {EntityAssetService, PreparedAssetUpload} from '@app/api/infrastructure/EntityAssetService'; import {Logger} from '@app/api/Logger'; import {UnknownGuildError} from '@fluxer/errors/src/domains/guild/UnknownGuildError'; @@ -21,6 +24,7 @@ interface AdminGuildUpdateServiceDeps { entityAssetService: EntityAssetService; auditService: AdminAuditService; updatePropagator: AdminGuildUpdatePropagator; + guildService: Pick; } export class AdminGuildUpdateService { @@ -39,23 +43,36 @@ export class AdminGuildUpdateService { adminUserId: UserID; auditLogReason: string | null; }) { - const {guildRepository, auditService, updatePropagator} = this.deps; - const guild = await guildRepository.findUnique(guildId); - if (!guild) { - throw new UnknownGuildError(); - } - const newFeatures = new Set(guild.features); - for (const feature of addFeatures) { - newFeatures.add(feature); - } - for (const feature of removeFeatures) { - newFeatures.delete(feature); - } - const updatedGuild = await guildRepository.upsertPartial(guildId, {features: newFeatures}, guild.toRow()); + const {guildRepository, auditService, updatePropagator, guildService} = this.deps; + const {guild, newFeatures, updatedGuild} = await guildService.withUpdateLock(guildId, async () => { + const guild = await guildRepository.findUnique(guildId); + if (!guild) { + throw new UnknownGuildError(); + } + const newFeatures = new Set(guild.features); + for (const feature of addFeatures) { + newFeatures.add(feature); + } + for (const feature of removeFeatures) { + newFeatures.delete(feature); + } + const patch: Partial = {features: newFeatures}; + if (staffEditTouchesInvitePause(guild, [...addFeatures, ...removeFeatures])) { + patch.invites_paused_until = null; + } + const updatedGuild = await guildRepository.upsertPartial(guildId, patch, guild.toRow()); + return {guild, newFeatures, updatedGuild}; + }); await updatePropagator.dispatchGuildUpdate(guildId, updatedGuild, { adminUserId, reconcileDiscoveryFeature: true, }); + await emitGuildFeaturesChanged({ + previous: guild, + updated: updatedGuild, + actorId: adminUserId, + source: 'staff', + }); await auditService.createAuditLog({ adminUserId, targetType: 'guild', diff --git a/fluxer_api/src/api/database/types/GuildTypes.ts b/fluxer_api/src/api/database/types/GuildTypes.ts index cd32ebd39..e62ba68a7 100644 --- a/fluxer_api/src/api/database/types/GuildTypes.ts +++ b/fluxer_api/src/api/database/types/GuildTypes.ts @@ -49,6 +49,7 @@ export interface GuildRow { audit_logs_indexed_at: Date | null; members_indexed_at: Date | null; message_history_cutoff: Date | null; + invites_paused_until: Date | null; version: number; } @@ -87,6 +88,7 @@ export const GUILD_COLUMNS = [ 'audit_logs_indexed_at', 'members_indexed_at', 'message_history_cutoff', + 'invites_paused_until', 'version', ] as const satisfies ReadonlyArray; diff --git a/fluxer_api/src/api/guild/GuildUpdateLock.ts b/fluxer_api/src/api/guild/GuildUpdateLock.ts new file mode 100644 index 000000000..8eddcb6e5 --- /dev/null +++ b/fluxer_api/src/api/guild/GuildUpdateLock.ts @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {GuildID} from '@app/api/BrandedTypes'; +import {ResourceLockedError} from '@fluxer/errors/src/domains/core/ResourceLockedError'; +import type {ICacheService} from '@pkgs/cache/src/ICacheService'; + +const GUILD_UPDATE_LOCK_TTL_SECONDS = 10; +const GUILD_UPDATE_LOCK_RETRY_DELAY_MS = 50; +const GUILD_UPDATE_LOCK_MAX_WAIT_MS = 5000; + +export type GuildUpdateLocks = Pick; + +async function acquireGuildUpdateLock(locks: GuildUpdateLocks, lockKey: string): Promise { + const startTime = Date.now(); + while (Date.now() - startTime < GUILD_UPDATE_LOCK_MAX_WAIT_MS) { + const token = await locks.acquireLock(lockKey, GUILD_UPDATE_LOCK_TTL_SECONDS); + if (token) { + return token; + } + await new Promise((resolve) => setTimeout(resolve, GUILD_UPDATE_LOCK_RETRY_DELAY_MS)); + } + return null; +} + +export async function withGuildUpdateLock( + locks: GuildUpdateLocks, + guildId: GuildID, + fn: () => Promise, +): Promise { + const lockKey = `guild:${guildId}:update`; + const lockToken = await acquireGuildUpdateLock(locks, lockKey); + if (!lockToken) { + throw new ResourceLockedError(); + } + try { + return await fn(); + } finally { + await locks.releaseLock(lockKey, lockToken); + } +} diff --git a/fluxer_api/src/api/guild/controllers/GuildDiscoveryController.ts b/fluxer_api/src/api/guild/controllers/GuildDiscoveryController.ts index b73f1a1a3..83725afda 100644 --- a/fluxer_api/src/api/guild/controllers/GuildDiscoveryController.ts +++ b/fluxer_api/src/api/guild/controllers/GuildDiscoveryController.ts @@ -4,6 +4,7 @@ import {createChannelID, createGuildID} from '@app/api/BrandedTypes'; import {Config} from '@app/api/Config'; import type {GuildDiscoveryRow} from '@app/api/database/types/GuildDiscoveryTypes'; import {viewerFromCtx} from '@app/api/experiment/ChannelThreadsGate'; +import {emitGuildJoinRefused} from '@app/api/infrastructure/activity/GuildEvents'; import {DefaultUserOnly, LoginRequired} from '@app/api/middleware/AuthMiddleware'; import {RateLimitMiddleware} from '@app/api/middleware/RateLimitMiddleware'; import {OpenAPI} from '@app/api/middleware/ResponseTypeMiddleware'; @@ -153,6 +154,9 @@ export function GuildDiscoveryController(app: HonoApp) { } const guild = await ctx.get('guildService').data.operations.getGuildSystem(guildId); if (guild.features.has(GuildFeatures.INVITES_DISABLED)) { + if (!(await ctx.get('gatewayService').hasGuildMember({guildId, userId: user.id}))) { + await emitGuildJoinRefused({guildId, userId: user.id, joinSource: 'discovery'}); + } throw new InvitesDisabledError(); } await ctx.get('guildService').members.addUserToGuild({ diff --git a/fluxer_api/src/api/guild/services/GuildInvitePauseApplier.ts b/fluxer_api/src/api/guild/services/GuildInvitePauseApplier.ts new file mode 100644 index 000000000..d02c66f2b --- /dev/null +++ b/fluxer_api/src/api/guild/services/GuildInvitePauseApplier.ts @@ -0,0 +1,160 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {createGuildID} from '@app/api/BrandedTypes'; +import {SYSTEM_USER_ID} from '@app/api/constants/Core'; +import type {GuildAuditLogService} from '@app/api/guild/GuildAuditLogService'; +import {mapGuildToGuildResponse} from '@app/api/guild/GuildModel'; +import {type GuildUpdateLocks, withGuildUpdateLock} from '@app/api/guild/GuildUpdateLock'; +import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; +import {workerMeta} from '@app/api/infrastructure/activity/ActivityMeta'; +import { + type ActionOutcome, + type FeatureSource, + INVITES_ALREADY_PAUSED, +} from '@app/api/infrastructure/activity/Contract.generated'; +import {emitGuildFeaturesChanged} from '@app/api/infrastructure/activity/GuildEvents'; +import type {IGatewayService} from '@app/api/infrastructure/IGatewayService'; +import {Logger} from '@app/api/Logger'; +import type {Guild} from '@app/api/models/Guild'; +import {type ActionOf, outcomeOf} from '@app/api/user/services/AccountStateApplier'; +import {serializeGuildForAudit} from '@app/api/utils/AuditSerializationUtils'; +import {AuditLogActionType} from '@fluxer/constants/src/AuditLogActionType'; +import {GuildFeatures} from '@fluxer/constants/src/GuildConstants'; +import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider'; + +export const GUILD_INVITE_PAUSE_SCHEDULE_KEY = 'guild:invite_pause:lifts'; +const INVITE_PAUSE_FEATURES = [GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED] as const; + +export interface GuildInvitePauseDeps { + guilds: Pick; + locks: GuildUpdateLocks; + schedule: Pick; + gateway: Pick; + auditLog: Pick; + now?: () => number; +} + +export function memberEndsInvitePause(guild: Guild, nextFeatures: ReadonlySet): boolean { + return ( + guild.invitesPausedUntil !== null && + guild.features.has(GuildFeatures.INVITES_DISABLED) && + !nextFeatures.has(GuildFeatures.INVITES_DISABLED) + ); +} + +export function staffEditTouchesInvitePause(guild: Guild, changed: Iterable): boolean { + if (guild.invitesPausedUntil === null) return false; + for (const feature of changed) { + if ((INVITE_PAUSE_FEATURES as ReadonlyArray).includes(feature)) return true; + } + return false; +} + +async function writePause( + deps: GuildInvitePauseDeps, + guild: Guild, + features: Set, + until: Date | null, + source: FeatureSource, +): Promise { + const updated = await deps.guilds.upsertPartial(guild.id, {features, invites_paused_until: until}, guild.toRow()); + try { + await deps.gateway.dispatchGuild({ + guildId: guild.id, + event: 'GUILD_UPDATE', + data: mapGuildToGuildResponse(updated), + }); + } catch (error) { + Logger.error({error, guildId: guild.id}, 'Failed to dispatch guild update after invite pause change'); + } + const changes = deps.auditLog + .computeChanges(serializeGuildForAudit(guild), serializeGuildForAudit(updated)) + .filter((change) => change.key === 'features'); + if (changes.length > 0) { + try { + await deps.auditLog.record({ + guildId: guild.id, + userId: SYSTEM_USER_ID, + action: AuditLogActionType.GUILD_UPDATE, + targetId: guild.id, + changes, + }); + } catch (error) { + Logger.error({error, guildId: guild.id}, 'Failed to record invite pause change in the guild audit log'); + } + } + await emitGuildFeaturesChanged({ + previous: guild, + updated, + actorId: null, + source, + meta: workerMeta(), + }); +} + +async function liftPause(deps: GuildInvitePauseDeps, guild: Guild, source: FeatureSource): Promise { + const features = new Set(guild.features); + for (const feature of INVITE_PAUSE_FEATURES) features.delete(feature); + await writePause(deps, guild, features, null, source); + await deps.schedule.zrem(GUILD_INVITE_PAUSE_SCHEDULE_KEY, guild.id.toString()); +} + +export async function applyPauseGuildInvites( + deps: GuildInvitePauseDeps, + env: ActionOf<'pause_guild_invites'>, +): Promise { + const guildId = createGuildID(BigInt(env.guild_id)); + return withGuildUpdateLock(deps.locks, guildId, async () => { + const guild = await deps.guilds.findUnique(guildId); + if (!guild) return outcomeOf(env, 'ineligible'); + const current = guild.invitesPausedUntil?.getTime() ?? null; + if (!env.on) { + if (current === null) return outcomeOf(env, 'noop'); + if (env.until_ms > 0 && current > env.until_ms) return outcomeOf(env, 'noop', null, `until_ms=${current}`); + await liftPause(deps, guild, 'action'); + return outcomeOf(env, 'applied'); + } + if (current === null && guild.features.has(GuildFeatures.INVITES_DISABLED)) { + return outcomeOf(env, 'noop', null, INVITES_ALREADY_PAUSED); + } + const now = deps.now?.() ?? Date.now(); + const until = Math.max(current ?? 0, env.until_ms); + if (until <= now) return outcomeOf(env, 'noop'); + const features = new Set(guild.features); + for (const feature of INVITE_PAUSE_FEATURES) features.add(feature); + const detail = `until_ms=${until}`; + await deps.schedule.zadd(GUILD_INVITE_PAUSE_SCHEDULE_KEY, until, guildId.toString()); + if (until === current && features.size === guild.features.size) return outcomeOf(env, 'noop', null, detail); + await writePause(deps, guild, features, new Date(until), 'action'); + return outcomeOf(env, 'applied', null, detail); + }); +} + +export async function liftExpiredGuildInvitePauses(deps: GuildInvitePauseDeps, limit: number): Promise { + const now = deps.now?.() ?? Date.now(); + const due = await deps.schedule.zrangebyscore(GUILD_INVITE_PAUSE_SCHEDULE_KEY, '-inf', now, 'LIMIT', 0, limit); + let lifted = 0; + for (const member of due) { + const guildId = createGuildID(BigInt(member)); + try { + const done = await withGuildUpdateLock(deps.locks, guildId, async () => { + const guild = await deps.guilds.findUnique(guildId); + const until = guild?.invitesPausedUntil?.getTime() ?? null; + if (!guild || until === null) { + await deps.schedule.zrem(GUILD_INVITE_PAUSE_SCHEDULE_KEY, member); + return false; + } + if (until > now) { + await deps.schedule.zadd(GUILD_INVITE_PAUSE_SCHEDULE_KEY, until, member); + return false; + } + await liftPause(deps, guild, 'system'); + return true; + }); + if (done) lifted++; + } catch (error) { + Logger.error({error, guildId: member}, 'Failed to lift expired guild invite pause'); + } + } + return lifted; +} diff --git a/fluxer_api/src/api/guild/services/GuildService.ts b/fluxer_api/src/api/guild/services/GuildService.ts index c0038a011..1557d6905 100644 --- a/fluxer_api/src/api/guild/services/GuildService.ts +++ b/fluxer_api/src/api/guild/services/GuildService.ts @@ -25,6 +25,7 @@ import { THREAD_AUDIT_LOG_ACTION_TYPES, } from '@app/api/guild/GuildAuditLogEntryMapper'; import type {GuildAuditLogService} from '@app/api/guild/GuildAuditLogService'; +import {withGuildUpdateLock} from '@app/api/guild/GuildUpdateLock'; import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; import {GuildChannelService} from '@app/api/guild/services/GuildChannelService'; import {GuildContentService} from '@app/api/guild/services/GuildContentService'; @@ -53,7 +54,6 @@ import {THREAD_FEATURE_CHANNEL_TYPES} from '@fluxer/constants/src/ThreadConstant import {ValidationErrorCodes} from '@fluxer/constants/src/ValidationErrorCodes'; import {InputValidationError} from '@fluxer/errors/src/domains/core/InputValidationError'; import {MissingPermissionsError} from '@fluxer/errors/src/domains/core/MissingPermissionsError'; -import {ResourceLockedError} from '@fluxer/errors/src/domains/core/ResourceLockedError'; import {UnknownGuildEmojiError} from '@fluxer/errors/src/domains/guild/UnknownGuildEmojiError'; import {UnknownGuildError} from '@fluxer/errors/src/domains/guild/UnknownGuildError'; import {UnknownGuildStickerError} from '@fluxer/errors/src/domains/guild/UnknownGuildStickerError'; @@ -80,10 +80,6 @@ interface StoredGuildAuditLogListResponse extends Omit; } -const GUILD_UPDATE_LOCK_TTL_SECONDS = 10; -const GUILD_UPDATE_LOCK_RETRY_DELAY_MS = 50; -const GUILD_UPDATE_LOCK_MAX_WAIT_MS = 5000; - export class GuildService { public readonly data: GuildDataService; public readonly members: GuildMemberService; @@ -211,9 +207,10 @@ export class GuildService { auditLogReason?: string | null, ): Promise { const {guildId, requestCache} = params; - const {guild, previousFeatures, updatedFeatures} = await this.withGuildUpdateLock(guildId, requestCache, () => - this.data.operations.updateGuild(params, auditLogReason), - ); + const {guild, previousFeatures, updatedFeatures} = await this.withUpdateLock(guildId, () => { + requestCache.guilds.delete(guildId); + return this.data.operations.updateGuild(params, auditLogReason); + }); if ( previousFeatures.has(GuildFeatures.TEXT_CHANNEL_FLEXIBLE_NAMES) && !updatedFeatures.has(GuildFeatures.TEXT_CHANNEL_FLEXIBLE_NAMES) @@ -223,30 +220,8 @@ export class GuildService { return guild; } - private async withGuildUpdateLock(guildId: GuildID, requestCache: RequestCache, fn: () => Promise): Promise { - const lockKey = `guild:${guildId}:update`; - const lockToken = await this.acquireGuildUpdateLock(lockKey); - if (!lockToken) { - throw new ResourceLockedError(); - } - try { - requestCache.guilds.delete(guildId); - return await fn(); - } finally { - await this.cacheService.releaseLock(lockKey, lockToken); - } - } - - private async acquireGuildUpdateLock(lockKey: string): Promise { - const startTime = Date.now(); - while (Date.now() - startTime < GUILD_UPDATE_LOCK_MAX_WAIT_MS) { - const token = await this.cacheService.acquireLock(lockKey, GUILD_UPDATE_LOCK_TTL_SECONDS); - if (token) { - return token; - } - await new Promise((resolve) => setTimeout(resolve, GUILD_UPDATE_LOCK_RETRY_DELAY_MS)); - } - return null; + withUpdateLock(guildId: GuildID, fn: () => Promise): Promise { + return withGuildUpdateLock(this.cacheService, guildId, fn); } async getEmojiMetadata(emojiId: EmojiID): Promise { diff --git a/fluxer_api/src/api/guild/services/data/GuildOperationsService.ts b/fluxer_api/src/api/guild/services/data/GuildOperationsService.ts index ff991ac42..d7ee5ee34 100644 --- a/fluxer_api/src/api/guild/services/data/GuildOperationsService.ts +++ b/fluxer_api/src/api/guild/services/data/GuildOperationsService.ts @@ -25,6 +25,8 @@ import {mapGuildToGuildResponse, mapGuildToPartialResponse} from '@app/api/guild import type {GuildDiscoveryRepository} from '@app/api/guild/repositories/GuildDiscoveryRepository'; import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; import type {GuildDataHelpers} from '@app/api/guild/services/data/GuildDataHelpers'; +import {memberEndsInvitePause} from '@app/api/guild/services/GuildInvitePauseApplier'; +import {emitGuildCreated, emitGuildFeaturesChanged} from '@app/api/infrastructure/activity/GuildEvents'; import {contentModerationService} from '@app/api/infrastructure/ContentModerationService'; import type {EntityAssetService, PreparedAssetUpload} from '@app/api/infrastructure/EntityAssetService'; import type {IGatewayService} from '@app/api/infrastructure/IGatewayService'; @@ -393,6 +395,7 @@ export class GuildOperationsService { audit_logs_indexed_at: null, members_indexed_at: null, message_history_cutoff: null, + invites_paused_until: null, version: 1, }; batch.addPrepared(Guilds.insert(guildData)); @@ -425,6 +428,7 @@ export class GuildOperationsService { await batch.execute(); await this.channelRepository.threads.ensureGuildMarker(guildId, {permsSeededAt: new Date()}); const guild = new Guild(guildData); + await emitGuildCreated(guild); await this.gatewayService.startGuild(guildId); await this.gatewayService.joinGuild({userId: user.id, guildId}); if (!user.isBot) { @@ -507,6 +511,10 @@ export class GuildOperationsService { } } const updatedFeatures = this.computeUpdatedFeatures(previousFeatures, data.features); + const endsInvitePause = memberEndsInvitePause(currentGuild, updatedFeatures); + if (endsInvitePause) { + updatedFeatures.delete(GuildFeatures.RAID_DETECTED); + } const featuresChanged = !setsEqual(previousFeatures, updatedFeatures); const patch: Partial = {}; if (data.name !== undefined) { @@ -617,6 +625,9 @@ export class GuildOperationsService { if (featuresChanged) { patch.features = updatedFeatures; } + if (endsInvitePause) { + patch.invites_paused_until = null; + } const preparedAssets = await this.prepareGuildAssetUpdates(currentGuild, data, patch); let updatedGuild: Guild; if (Object.keys(patch).length === 0) { @@ -635,6 +646,14 @@ export class GuildOperationsService { Logger.error({error, guildId}, 'Failed to commit asset changes after successful guild update'); } await this.helpers.dispatchGuildUpdate(updatedGuild); + if (featuresChanged) { + await emitGuildFeaturesChanged({ + previous: currentGuild, + updated: updatedGuild, + actorId: userId, + source: 'member', + }); + } const guildSearchService = getGuildSearchService(); if (guildSearchService) { let discoveryContext: GuildDiscoveryContext | undefined; diff --git a/fluxer_api/src/api/guild/services/member/GuildMemberOperationsService.ts b/fluxer_api/src/api/guild/services/member/GuildMemberOperationsService.ts index 2a91c4640..7bc9a776b 100644 --- a/fluxer_api/src/api/guild/services/member/GuildMemberOperationsService.ts +++ b/fluxer_api/src/api/guild/services/member/GuildMemberOperationsService.ts @@ -23,6 +23,7 @@ import type {GuildMemberSearchIndexService} from '@app/api/guild/services/member import type {GuildMemberValidationService} from '@app/api/guild/services/member/GuildMemberValidationService'; import {emitActivity} from '@app/api/infrastructure/activity/ActivityEvents'; import type {JoinSource} from '@app/api/infrastructure/activity/Contract.generated'; +import {emitGuildMemberRemoved} from '@app/api/infrastructure/activity/GuildEvents'; import {contentModerationService} from '@app/api/infrastructure/ContentModerationService'; import type {EntityAssetService, PreparedAssetUpload} from '@app/api/infrastructure/EntityAssetService'; import type {IGatewayService} from '@app/api/infrastructure/IGatewayService'; @@ -388,6 +389,7 @@ export class GuildMemberOperationsService { ); } await this.gatewayService.leaveGuild({userId: targetId, guildId}); + await emitGuildMemberRemoved({guildId, userId: targetId, actorId: userId, kind: 'kicked'}); } async addUserToGuild( @@ -541,6 +543,7 @@ export class GuildMemberOperationsService { }); } await this.gatewayService.leaveGuild({userId, guildId}); + await emitGuildMemberRemoved({guildId, userId, actorId: null, kind: 'left'}); } private async validateTimeout(params: { diff --git a/fluxer_api/src/api/guild/tests/DiscoverySearchAndJoin.test.ts b/fluxer_api/src/api/guild/tests/DiscoverySearchAndJoin.test.ts index cd5fd436f..13627a79c 100644 --- a/fluxer_api/src/api/guild/tests/DiscoverySearchAndJoin.test.ts +++ b/fluxer_api/src/api/guild/tests/DiscoverySearchAndJoin.test.ts @@ -6,7 +6,14 @@ import {createTestBotAccount} from '@app/api/bot/tests/BotTestUtils'; import {createPermissionOverwrite} from '@app/api/channel/tests/ChannelTestUtils'; import {threadsRequest} from '@app/api/channel/tests/ThreadTestUtils'; import {clearChannelThreadsTaintCacheForTesting} from '@app/api/experiment/ChannelThreadsGate'; -import {createChannel, createGuild} from '@app/api/guild/tests/GuildTestUtils'; +import {createChannel, createGuild, updateGuild} from '@app/api/guild/tests/GuildTestUtils'; +import { + idleActivityEvents, + resetActivityEventsForTests, + startActivityEvents, +} from '@app/api/infrastructure/activity/ActivityEvents'; +import type {ActivityPublisher} from '@app/api/infrastructure/activity/ActivitySpool'; +import type {Event} from '@app/api/infrastructure/activity/Contract.generated'; import {setInjectedGatewayService} from '@app/api/middleware/ServiceRegistry'; import {getGuildRepository} from '@app/api/middleware/ServiceSingletons'; import {banUser} from '@app/api/moderation/tests/ModerationTestUtils'; @@ -20,11 +27,13 @@ import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/wo import {APIErrorCodes} from '@fluxer/constants/src/ApiErrorCodes'; import {ChannelTypes, Permissions} from '@fluxer/constants/src/ChannelConstants'; import {DiscoveryCategories} from '@fluxer/constants/src/DiscoveryConstants'; +import {GuildFeatures} from '@fluxer/constants/src/GuildConstants'; import type { DiscoveryApplicationResponse, DiscoveryChannelPreviewResponse, DiscoveryGuildListResponse, } from '@fluxer/schema/src/domains/guild/GuildDiscoverySchemas'; +import {MockKVProvider} from '@pkgs/kv_client/src/testing/MockKVProvider'; import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask'; import {afterEach, beforeEach, describe, expect, test} from 'vitest'; @@ -42,6 +51,14 @@ interface LiveGuildCounts { const WORKER_HELPERS = {logger: new NoopLogger()} as unknown as WorkerTaskHelpers; +class CapturingPublisher implements ActivityPublisher { + readonly events: Array = []; + + async publish(_subject: string, payload: string): Promise { + this.events.push(JSON.parse(payload) as Event); + } +} + class LiveCountsGatewayService extends NoopGatewayService { constructor(private readonly liveCounts: Map) { super(); @@ -248,6 +265,46 @@ describe('Discovery Search and Join', () => { .expect(HTTP_STATUS.BAD_REQUEST, APIErrorCodes.DISCOVERY_NOT_DISCOVERABLE) .execute(); }); + test('should publish a refusal for a non-member discovery join while invites are disabled', async () => { + const publisher = new CapturingPublisher(); + await startActivityEvents({publisher, kv: new MockKVProvider()}); + try { + const owner = await createTestAccount(harness); + const guild = await createGuild(harness, owner.token, 'Closed Discovery Guild'); + await setGuildMemberCount(harness, guild.id, 10); + const admin = await createTestAccount(harness); + await setUserACLs(harness, admin, ['admin:authenticate', 'discovery:review']); + await applyAndApprove( + harness, + owner.token, + admin.token, + guild.id, + 'Join this closed community', + DiscoveryCategories.GAMING, + ); + await updateGuild(harness, owner.token, guild.id, { + features: [...guild.features, GuildFeatures.INVITES_DISABLED], + }); + const joiner = await createTestAccount(harness); + for (const account of [owner, joiner]) { + await createBuilder(harness, account.token) + .post(`/discovery/guilds/${guild.id}/join`) + .expect(HTTP_STATUS.FORBIDDEN, APIErrorCodes.INVITES_DISABLED) + .execute(); + } + await idleActivityEvents(); + const refusals = publisher.events.filter((event) => event.kind === 'guild_join_refused'); + expect(refusals.map((event) => [event.key, event.data])).toEqual([ + [ + joiner.userId, + {guild_id: guild.id, user_id: joiner.userId, cause: 'invites_paused', join_source: 'discovery'}, + ], + ]); + } finally { + resetActivityEventsForTests(); + } + }); + test('should not allow bot accounts to join discoverable guilds', async () => { const owner = await createTestAccount(harness); const guild = await createGuild(harness, owner.token, 'No Bots Guild'); diff --git a/fluxer_api/src/api/guild/tests/GuildInvitePauseApplier.test.ts b/fluxer_api/src/api/guild/tests/GuildInvitePauseApplier.test.ts new file mode 100644 index 000000000..c871fe1a5 --- /dev/null +++ b/fluxer_api/src/api/guild/tests/GuildInvitePauseApplier.test.ts @@ -0,0 +1,403 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {createGuildID, createUserID, type GuildID} from '@app/api/BrandedTypes'; +import {SYSTEM_USER_ID} from '@app/api/constants/Core'; +import type {GuildRow} from '@app/api/database/types/GuildTypes'; +import {GuildAuditLogService} from '@app/api/guild/GuildAuditLogService'; +import { + applyPauseGuildInvites, + GUILD_INVITE_PAUSE_SCHEDULE_KEY, + type GuildInvitePauseDeps, + liftExpiredGuildInvitePauses, +} from '@app/api/guild/services/GuildInvitePauseApplier'; +import {resetActivityEventsForTests, startActivityEvents} from '@app/api/infrastructure/activity/ActivityEvents'; +import type {ActivityPublisher} from '@app/api/infrastructure/activity/ActivitySpool'; +import type {ActionEnvelope, Event} from '@app/api/infrastructure/activity/Contract.generated'; +import {Guild} from '@app/api/models/Guild'; +import {applyAction} from '@app/api/worker/AccountActionConsumer'; +import {AuditLogActionType} from '@fluxer/constants/src/AuditLogActionType'; +import {GuildFeatures} from '@fluxer/constants/src/GuildConstants'; +import {MockKVProvider} from '@pkgs/kv_client/src/testing/MockKVProvider'; +import {afterEach, beforeEach, describe, expect, it} from 'vitest'; + +const GUILD_ID = '1472623911696138261'; +const NOW = 1_760_000_000_000; +const HOUR_MS = 3_600_000; +const BASE_FEATURES = [GuildFeatures.BANNER]; + +type PauseEnvelope = Extract; + +function pauseEnvelope(overrides: Partial = {}): PauseEnvelope { + return { + v: 1, + id: 'a:11:guild:0', + key: `guild:${GUILD_ID}`, + issued_at_ms: NOW, + expires_at_ms: NOW + 60_000, + type: 'pause_guild_invites', + guild_id: GUILD_ID, + on: true, + until_ms: NOW + 2 * HOUR_MS, + ...overrides, + }; +} + +function guildRow(features: Iterable, pausedUntil: Date | null = null): GuildRow { + return { + guild_id: createGuildID(BigInt(GUILD_ID)), + owner_id: createUserID(1472623911696138262n), + name: 'Paused', + vanity_url_code: null, + icon_hash: null, + banner_hash: null, + banner_width: null, + banner_height: null, + splash_hash: null, + splash_width: null, + splash_height: null, + splash_card_alignment: null, + embed_splash_hash: null, + embed_splash_width: null, + embed_splash_height: null, + features: new Set(features), + verification_level: 0, + mfa_level: 0, + nsfw_level: 0, + explicit_content_filter: 0, + default_message_notifications: 0, + system_channel_id: null, + system_channel_flags: 0, + rules_channel_id: null, + afk_channel_id: null, + afk_timeout: 0, + disabled_operations: 0, + member_count: 12, + audit_logs_indexed_at: null, + members_indexed_at: null, + message_history_cutoff: null, + invites_paused_until: pausedUntil, + version: 1, + }; +} + +class FakeGuilds { + row: GuildRow | null = null; + writes = 0; + + current(): Guild { + return new Guild(this.row!); + } + + async findUnique(_guildId: GuildID): Promise { + return this.row ? new Guild(this.row) : null; + } + + async upsertPartial(_guildId: GuildID, patch: Partial): Promise { + this.writes++; + this.row = {...this.row!, ...patch, version: this.row!.version + 1}; + return new Guild(this.row); + } +} + +class CapturingPublisher implements ActivityPublisher { + readonly events: Array = []; + + async publish(_subject: string, payload: string): Promise { + this.events.push(JSON.parse(payload) as Event); + } + + featureChanges() { + return this.events.flatMap((event) => (event.kind === 'guild_features_changed' ? [event] : [])); + } +} + +interface Harness { + guilds: FakeGuilds; + schedule: MockKVProvider; + dispatched: Array; + audits: Array<{userId: bigint; action: number; changes: unknown}>; + clock: {now: number}; + deps: GuildInvitePauseDeps; +} + +function harness(): Harness { + const guilds = new FakeGuilds(); + const schedule = new MockKVProvider(); + const h: Harness = {guilds, schedule, dispatched: [], audits: [], clock: {now: NOW}, deps: null as never}; + const auditService = new GuildAuditLogService(null as never, null as never, null as never, null as never); + h.deps = { + guilds, + locks: {acquireLock: async () => 'token', releaseLock: async () => true}, + schedule, + gateway: { + dispatchGuild: async (params) => { + h.dispatched.push(params.data); + }, + }, + auditLog: { + computeChanges: (previous, next) => auditService.computeChanges(previous, next), + record: async (params) => { + h.audits.push({userId: params.userId, action: params.action, changes: params.changes}); + }, + }, + now: () => h.clock.now, + }; + return h; +} + +async function scheduled(h: Harness): Promise> { + return h.schedule.zrangebyscore(GUILD_INVITE_PAUSE_SCHEDULE_KEY, '-inf', '+inf'); +} + +describe('guild invite pause', () => { + let h: Harness; + let publisher: CapturingPublisher; + + beforeEach(async () => { + h = harness(); + publisher = new CapturingPublisher(); + await startActivityEvents({publisher, kv: new MockKVProvider()}); + }); + + afterEach(() => { + resetActivityEventsForTests(); + }); + + it('leaves a pause the guild set itself alone and reports it as already paused', async () => { + h.guilds.row = guildRow([...BASE_FEATURES, GuildFeatures.INVITES_DISABLED]); + const outcome = await applyPauseGuildInvites(h.deps, pauseEnvelope()); + expect(outcome).toEqual({ + action_id: 'a:11:guild:0', + action_type: 'pause_guild_invites', + status: 'noop', + detail: 'already_paused', + observed: null, + guild_id: GUILD_ID, + }); + expect(h.guilds.writes).toBe(0); + expect(await scheduled(h)).toEqual([]); + expect(publisher.events).toEqual([]); + const lifted = await applyPauseGuildInvites(h.deps, pauseEnvelope({on: false})); + expect(lifted.status).toBe('noop'); + expect(h.guilds.current().features.has(GuildFeatures.INVITES_DISABLED)).toBe(true); + }); + + it('pauses invites, records it as the system user and schedules the lift', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + const outcome = await applyPauseGuildInvites(h.deps, pauseEnvelope()); + expect(outcome).toMatchObject({status: 'applied', detail: `until_ms=${NOW + 2 * HOUR_MS}`, guild_id: GUILD_ID}); + const guild = h.guilds.current(); + expect([...guild.features].sort()).toEqual( + [...BASE_FEATURES, GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED].sort(), + ); + expect(guild.invitesPausedUntil?.getTime()).toBe(NOW + 2 * HOUR_MS); + expect(h.dispatched).toHaveLength(1); + expect(h.audits).toEqual([ + { + userId: SYSTEM_USER_ID, + action: AuditLogActionType.GUILD_UPDATE, + changes: [ + { + key: 'features', + old_value: [...BASE_FEATURES], + new_value: [...BASE_FEATURES, GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED].sort(), + }, + ], + }, + ]); + expect(await scheduled(h)).toEqual([GUILD_ID]); + const [event] = publisher.featureChanges(); + expect(event?.key).toBe(`guild:${GUILD_ID}`); + expect(event?.meta.channel).toBe('worker'); + expect(event?.data).toEqual({ + guild_id: GUILD_ID, + actor_id: null, + source: 'action', + added: [GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED], + removed: [], + invites_paused_until_ms: NOW + 2 * HOUR_MS, + }); + }); + + it('extends its own pause but never shortens it', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + await applyPauseGuildInvites(h.deps, pauseEnvelope()); + const extended = await applyPauseGuildInvites( + h.deps, + pauseEnvelope({id: 'a:11:guild:1', until_ms: NOW + 3 * HOUR_MS}), + ); + expect(extended.status).toBe('applied'); + expect(h.guilds.current().invitesPausedUntil?.getTime()).toBe(NOW + 3 * HOUR_MS); + expect( + await h.schedule.zrangebyscore(GUILD_INVITE_PAUSE_SCHEDULE_KEY, NOW + 3 * HOUR_MS, NOW + 3 * HOUR_MS), + ).toEqual([GUILD_ID]); + const shorter = await applyPauseGuildInvites(h.deps, pauseEnvelope({id: 'a:11:guild:2', until_ms: NOW + HOUR_MS})); + expect(shorter).toMatchObject({status: 'noop', detail: `until_ms=${NOW + 3 * HOUR_MS}`}); + expect(h.guilds.current().invitesPausedUntil?.getTime()).toBe(NOW + 3 * HOUR_MS); + expect(publisher.featureChanges().map((event) => event.data.invites_paused_until_ms)).toEqual([ + NOW + 2 * HOUR_MS, + NOW + 3 * HOUR_MS, + ]); + }); + + it('lifts only its own pause and removes both features', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + await applyPauseGuildInvites(h.deps, pauseEnvelope()); + const outcome = await applyAction( + {js: {} as never, state: {} as never, guildInvites: h.deps, now: () => NOW}, + pauseEnvelope({id: 'a:11:guild:3', on: false}), + ); + expect(outcome).toMatchObject({status: 'applied', guild_id: GUILD_ID}); + const guild = h.guilds.current(); + expect([...guild.features]).toEqual(BASE_FEATURES); + expect(guild.invitesPausedUntil).toBeNull(); + expect(await scheduled(h)).toEqual([]); + expect(publisher.featureChanges().at(-1)?.data).toMatchObject({ + source: 'action', + removed: [GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED], + invites_paused_until_ms: null, + }); + const again = await applyPauseGuildInvites(h.deps, pauseEnvelope({id: 'a:11:guild:4', on: false})); + expect(again.status).toBe('noop'); + }); + + it('ignores a late lift for an earlier window once the pause was extended', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + await applyPauseGuildInvites(h.deps, pauseEnvelope()); + await applyPauseGuildInvites(h.deps, pauseEnvelope({id: 'a:11:guild:5', until_ms: NOW + 3 * HOUR_MS})); + const late = await applyPauseGuildInvites( + h.deps, + pauseEnvelope({id: 'a:11:guild:6', on: false, until_ms: NOW + 2 * HOUR_MS}), + ); + expect(late).toMatchObject({status: 'noop', detail: `until_ms=${NOW + 3 * HOUR_MS}`}); + expect(h.guilds.current().features.has(GuildFeatures.INVITES_DISABLED)).toBe(true); + expect(h.guilds.current().invitesPausedUntil?.getTime()).toBe(NOW + 3 * HOUR_MS); + const current = await applyPauseGuildInvites( + h.deps, + pauseEnvelope({id: 'a:11:guild:7', on: false, until_ms: NOW + 3 * HOUR_MS}), + ); + expect(current.status).toBe('applied'); + expect(h.guilds.current().invitesPausedUntil).toBeNull(); + }); + + it('keeps the lift scheduled when the gateway fails after the write and a redelivery changes nothing', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + const dispatch = h.deps.gateway.dispatchGuild; + let failures = 1; + h.deps.gateway = { + dispatchGuild: async (params) => { + if (failures-- > 0) throw new Error('gateway timeout'); + await dispatch(params); + }, + }; + const first = await applyPauseGuildInvites(h.deps, pauseEnvelope()); + expect(first).toMatchObject({status: 'applied', guild_id: GUILD_ID}); + expect(h.audits).toHaveLength(1); + expect(publisher.featureChanges()).toHaveLength(1); + const redelivered = await applyPauseGuildInvites(h.deps, pauseEnvelope()); + expect(redelivered).toMatchObject({status: 'noop', detail: `until_ms=${NOW + 2 * HOUR_MS}`}); + expect(h.guilds.writes).toBe(1); + expect(await scheduled(h)).toEqual([GUILD_ID]); + h.clock.now = NOW + 2 * HOUR_MS; + expect(await liftExpiredGuildInvitePauses(h.deps, 100)).toBe(1); + expect([...h.guilds.current().features]).toEqual(BASE_FEATURES); + }); + + it('restores a lost schedule entry when the same pause is redelivered', async () => { + h.guilds.row = guildRow( + [...BASE_FEATURES, GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED], + new Date(NOW + 2 * HOUR_MS), + ); + const outcome = await applyPauseGuildInvites(h.deps, pauseEnvelope()); + expect(outcome.status).toBe('noop'); + expect(h.guilds.writes).toBe(0); + expect(await scheduled(h)).toEqual([GUILD_ID]); + }); + + it('writes nothing when scheduling fails so the retry applies the pause', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + const zadd = h.schedule.zadd.bind(h.schedule); + let failures = 1; + h.deps.schedule = { + zadd: async (...args: Parameters) => { + if (failures-- > 0) throw new Error('kv unavailable'); + return zadd(...args); + }, + zrem: h.schedule.zrem.bind(h.schedule), + zrangebyscore: h.schedule.zrangebyscore.bind(h.schedule), + }; + await expect(applyPauseGuildInvites(h.deps, pauseEnvelope())).rejects.toThrow('kv unavailable'); + expect(h.guilds.writes).toBe(0); + expect((await applyPauseGuildInvites(h.deps, pauseEnvelope())).status).toBe('applied'); + expect(await scheduled(h)).toEqual([GUILD_ID]); + }); + + it('routes the action through the consumer and answers every outcome with the guild', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + const consumer = {js: {} as never, state: {} as never, guildInvites: h.deps, now: () => NOW}; + expect(await applyAction(consumer, pauseEnvelope())).toMatchObject({status: 'applied', guild_id: GUILD_ID}); + expect(await applyAction(consumer, pauseEnvelope({expires_at_ms: NOW}))).toMatchObject({ + status: 'expired', + guild_id: GUILD_ID, + }); + expect(await applyAction(consumer, {...pauseEnvelope(), v: 2})).toMatchObject({ + status: 'unsupported', + guild_id: GUILD_ID, + }); + h.guilds.row = guildRow([...BASE_FEATURES, GuildFeatures.INVITES_DISABLED]); + expect(await applyAction(consumer, pauseEnvelope({id: 'a:11:guild:5'}))).toMatchObject({ + status: 'noop', + detail: 'already_paused', + guild_id: GUILD_ID, + }); + expect(h.guilds.writes).toBe(1); + }); + + it('removes a notice staff set before the pause when the pause ends', async () => { + h.guilds.row = guildRow([...BASE_FEATURES, GuildFeatures.RAID_DETECTED]); + expect((await applyPauseGuildInvites(h.deps, pauseEnvelope())).status).toBe('applied'); + h.clock.now = NOW + 2 * HOUR_MS; + expect(await liftExpiredGuildInvitePauses(h.deps, 100)).toBe(1); + expect([...h.guilds.current().features]).toEqual(BASE_FEATURES); + }); + + it('reports a missing guild as ineligible', async () => { + expect((await applyPauseGuildInvites(h.deps, pauseEnvelope())).status).toBe('ineligible'); + }); + + it('lifts an expired pause on the sweep without any action', async () => { + h.guilds.row = guildRow(BASE_FEATURES); + await applyPauseGuildInvites(h.deps, pauseEnvelope()); + expect(await liftExpiredGuildInvitePauses(h.deps, 100)).toBe(0); + expect(h.guilds.current().features.has(GuildFeatures.INVITES_DISABLED)).toBe(true); + h.clock.now = NOW + 2 * HOUR_MS; + expect(await liftExpiredGuildInvitePauses(h.deps, 100)).toBe(1); + const guild = h.guilds.current(); + expect([...guild.features]).toEqual(BASE_FEATURES); + expect(guild.invitesPausedUntil).toBeNull(); + expect(await scheduled(h)).toEqual([]); + expect(publisher.featureChanges().at(-1)?.data).toMatchObject({ + source: 'system', + actor_id: null, + removed: [GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED], + }); + }); + + it('drops a sweep entry once the pause is no longer ours and reschedules an extended one', async () => { + h.guilds.row = guildRow([...BASE_FEATURES, GuildFeatures.INVITES_DISABLED]); + await h.schedule.zadd(GUILD_INVITE_PAUSE_SCHEDULE_KEY, NOW - 1, GUILD_ID); + expect(await liftExpiredGuildInvitePauses(h.deps, 100)).toBe(0); + expect(await scheduled(h)).toEqual([]); + expect(h.guilds.current().features.has(GuildFeatures.INVITES_DISABLED)).toBe(true); + h.guilds.row = guildRow( + [...BASE_FEATURES, GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED], + new Date(NOW + HOUR_MS), + ); + await h.schedule.zadd(GUILD_INVITE_PAUSE_SCHEDULE_KEY, NOW - 1, GUILD_ID); + expect(await liftExpiredGuildInvitePauses(h.deps, 100)).toBe(0); + expect(await h.schedule.zrangebyscore(GUILD_INVITE_PAUSE_SCHEDULE_KEY, NOW + HOUR_MS, NOW + HOUR_MS)).toEqual([ + GUILD_ID, + ]); + expect(h.guilds.writes).toBe(0); + }); +}); diff --git a/fluxer_api/src/api/guild/tests/GuildModel.test.ts b/fluxer_api/src/api/guild/tests/GuildModel.test.ts index e5e61aa66..df92200a4 100644 --- a/fluxer_api/src/api/guild/tests/GuildModel.test.ts +++ b/fluxer_api/src/api/guild/tests/GuildModel.test.ts @@ -43,6 +43,7 @@ function createGuild(overrides: Partial = {}): Guild { audit_logs_indexed_at: null, members_indexed_at: null, message_history_cutoff: null, + invites_paused_until: null, version: 1, ...overrides, }); diff --git a/fluxer_api/src/api/infrastructure/activity/Contract.generated.ts b/fluxer_api/src/api/infrastructure/activity/Contract.generated.ts index 14deb4961..9e1a62e2e 100644 --- a/fluxer_api/src/api/infrastructure/activity/Contract.generated.ts +++ b/fluxer_api/src/api/infrastructure/activity/Contract.generated.ts @@ -4,9 +4,9 @@ export type Id = string; export type Flags64 = string; export type Channel = "stable" | "canary" | "worker" | "internal" | "import" | "other"; export type Meta = { ip: string | null, country: string | null, ua: string | null, locale: string | null, channel: Channel, request_id: string | null, }; -export type Kind = "registration" | "email_changed" | "profile_updated" | "account_changed" | "admin_action" | "account_deleted" | "report_filed" | "email_bounced" | "action_outcome" | "login" | "session_started" | "guild_joined" | "dm_opened" | "message_created" | "message_updated" | "friend_request" | "http_errors" | "user_blocked" | "guild_member_banned" | "guild_member_unbanned" | "report_resolved"; -export type Event = { v: number, id: string, at_ms: number, key: string, meta: Meta, } & ({ "kind": "registration", "data": Registration } | { "kind": "email_changed", "data": EmailChanged } | { "kind": "profile_updated", "data": ProfileUpdated } | { "kind": "account_changed", "data": AccountChanged } | { "kind": "admin_action", "data": AdminAction } | { "kind": "account_deleted", "data": AccountDeleted } | { "kind": "report_filed", "data": ReportFiled } | { "kind": "email_bounced", "data": EmailBounced } | { "kind": "action_outcome", "data": ActionOutcome } | { "kind": "login", "data": Login } | { "kind": "session_started", "data": SessionStarted } | { "kind": "guild_joined", "data": GuildJoined } | { "kind": "dm_opened", "data": DmOpened } | { "kind": "message_created", "data": MessageCreated } | { "kind": "message_updated", "data": MessageUpdated } | { "kind": "friend_request", "data": FriendRequest } | { "kind": "http_errors", "data": HttpErrors } | { "kind": "user_blocked", "data": UserBlocked } | { "kind": "guild_member_banned", "data": GuildMemberBanned } | { "kind": "guild_member_unbanned", "data": GuildMemberUnbanned } | { "kind": "report_resolved", "data": ReportResolved }); -export type Body = { "kind": "registration", "data": Registration } | { "kind": "email_changed", "data": EmailChanged } | { "kind": "profile_updated", "data": ProfileUpdated } | { "kind": "account_changed", "data": AccountChanged } | { "kind": "admin_action", "data": AdminAction } | { "kind": "account_deleted", "data": AccountDeleted } | { "kind": "report_filed", "data": ReportFiled } | { "kind": "email_bounced", "data": EmailBounced } | { "kind": "action_outcome", "data": ActionOutcome } | { "kind": "login", "data": Login } | { "kind": "session_started", "data": SessionStarted } | { "kind": "guild_joined", "data": GuildJoined } | { "kind": "dm_opened", "data": DmOpened } | { "kind": "message_created", "data": MessageCreated } | { "kind": "message_updated", "data": MessageUpdated } | { "kind": "friend_request", "data": FriendRequest } | { "kind": "http_errors", "data": HttpErrors } | { "kind": "user_blocked", "data": UserBlocked } | { "kind": "guild_member_banned", "data": GuildMemberBanned } | { "kind": "guild_member_unbanned", "data": GuildMemberUnbanned } | { "kind": "report_resolved", "data": ReportResolved }; +export type Kind = "registration" | "email_changed" | "profile_updated" | "account_changed" | "admin_action" | "account_deleted" | "report_filed" | "email_bounced" | "action_outcome" | "login" | "session_started" | "guild_joined" | "dm_opened" | "message_created" | "message_updated" | "friend_request" | "http_errors" | "user_blocked" | "guild_member_banned" | "guild_member_unbanned" | "report_resolved" | "guild_created" | "guild_features_changed" | "invite_created" | "invite_deleted" | "guild_member_removed" | "guild_join_refused"; +export type Event = { v: number, id: string, at_ms: number, key: string, meta: Meta, } & ({ "kind": "registration", "data": Registration } | { "kind": "email_changed", "data": EmailChanged } | { "kind": "profile_updated", "data": ProfileUpdated } | { "kind": "account_changed", "data": AccountChanged } | { "kind": "admin_action", "data": AdminAction } | { "kind": "account_deleted", "data": AccountDeleted } | { "kind": "report_filed", "data": ReportFiled } | { "kind": "email_bounced", "data": EmailBounced } | { "kind": "action_outcome", "data": ActionOutcome } | { "kind": "login", "data": Login } | { "kind": "session_started", "data": SessionStarted } | { "kind": "guild_joined", "data": GuildJoined } | { "kind": "dm_opened", "data": DmOpened } | { "kind": "message_created", "data": MessageCreated } | { "kind": "message_updated", "data": MessageUpdated } | { "kind": "friend_request", "data": FriendRequest } | { "kind": "http_errors", "data": HttpErrors } | { "kind": "user_blocked", "data": UserBlocked } | { "kind": "guild_member_banned", "data": GuildMemberBanned } | { "kind": "guild_member_unbanned", "data": GuildMemberUnbanned } | { "kind": "report_resolved", "data": ReportResolved } | { "kind": "guild_created", "data": GuildCreated } | { "kind": "guild_features_changed", "data": GuildFeaturesChanged } | { "kind": "invite_created", "data": InviteCreated } | { "kind": "invite_deleted", "data": InviteDeleted } | { "kind": "guild_member_removed", "data": GuildMemberRemoved } | { "kind": "guild_join_refused", "data": GuildJoinRefused }); +export type Body = { "kind": "registration", "data": Registration } | { "kind": "email_changed", "data": EmailChanged } | { "kind": "profile_updated", "data": ProfileUpdated } | { "kind": "account_changed", "data": AccountChanged } | { "kind": "admin_action", "data": AdminAction } | { "kind": "account_deleted", "data": AccountDeleted } | { "kind": "report_filed", "data": ReportFiled } | { "kind": "email_bounced", "data": EmailBounced } | { "kind": "action_outcome", "data": ActionOutcome } | { "kind": "login", "data": Login } | { "kind": "session_started", "data": SessionStarted } | { "kind": "guild_joined", "data": GuildJoined } | { "kind": "dm_opened", "data": DmOpened } | { "kind": "message_created", "data": MessageCreated } | { "kind": "message_updated", "data": MessageUpdated } | { "kind": "friend_request", "data": FriendRequest } | { "kind": "http_errors", "data": HttpErrors } | { "kind": "user_blocked", "data": UserBlocked } | { "kind": "guild_member_banned", "data": GuildMemberBanned } | { "kind": "guild_member_unbanned", "data": GuildMemberUnbanned } | { "kind": "report_resolved", "data": ReportResolved } | { "kind": "guild_created", "data": GuildCreated } | { "kind": "guild_features_changed", "data": GuildFeaturesChanged } | { "kind": "invite_created", "data": InviteCreated } | { "kind": "invite_deleted", "data": InviteDeleted } | { "kind": "guild_member_removed", "data": GuildMemberRemoved } | { "kind": "guild_join_refused", "data": GuildJoinRefused }; export type Registration = { user_id: Id, method: RegMethod, email: string | null, username: string, username_user_chosen: boolean, global_name: string | null, locale: string | null, timezone: string | null, invite_code: string | null, flags: Flags64, date_of_birth: string | null, }; export type RegMethod = "password" | "unclaimed" | "oauth" | "other"; export type EmailChanged = { user_id: Id, new_email: string, was_unclaimed: boolean, has_ever_purchased: boolean, }; @@ -38,9 +38,18 @@ export type BanBy = "moderator" | "staff"; export type ReportResolved = { report_id: Id, reporter_id: Id, category: string, target_type: ReportTarget, reported_user_id: Id | null, outcome: ReportOutcome, resolved_by: ResolvedBy, }; export type ReportOutcome = "actioned" | "no_violation" | "duplicate" | "auto_resolved" | "unspecified"; export type ResolvedBy = "staff" | "system"; -export type ActionEnvelope = { v: number, id: string, key: string, issued_at_ms: number, expires_at_ms: number, } & ({ "type": "set_account_limit", user_id: Id, on: boolean, } | { "type": "temp_ban_ip", ip: string, until_ms: number, } | { "type": "limit_new_conversations", user_id: Id, on: boolean, until_ms: number, } | { "type": "hide_recent_messages", user_id: Id, since_ms: number, on: boolean, } | { "type": "hide_profile", user_id: Id, on: boolean, } | { "type": "delete_user_messages", user_id: Id, on: boolean, }); -export type Action = { "type": "set_account_limit", user_id: Id, on: boolean, } | { "type": "temp_ban_ip", ip: string, until_ms: number, } | { "type": "limit_new_conversations", user_id: Id, on: boolean, until_ms: number, } | { "type": "hide_recent_messages", user_id: Id, since_ms: number, on: boolean, } | { "type": "hide_profile", user_id: Id, on: boolean, } | { "type": "delete_user_messages", user_id: Id, on: boolean, }; -export type ActionOutcome = { action_id: string, action_type: string, status: OutcomeStatus, detail: string | null, observed: Observed | null, user_id?: Id, }; +export type GuildCreated = { guild_id: Id, owner_id: Id, member_count: number | null, }; +export type GuildFeaturesChanged = { guild_id: Id, actor_id: Id | null, source: FeatureSource, added: Array, removed: Array, invites_paused_until_ms: number | null, }; +export type FeatureSource = "member" | "staff" | "action" | "system"; +export type InviteCreated = { guild_id: Id, channel_id: Id | null, code: string, inviter_id: Id | null, max_uses: number, max_age_s: number, temporary: boolean, }; +export type InviteDeleted = { guild_id: Id, code: string, }; +export type GuildMemberRemoved = { guild_id: Id, user_id: Id, actor_id: Id | null, kind: RemovalKind, }; +export type RemovalKind = "left" | "kicked"; +export type GuildJoinRefused = { guild_id: Id, user_id: Id | null, cause: RefusalCause, join_source: JoinSource, }; +export type RefusalCause = "invites_paused" | "other"; +export type ActionEnvelope = { v: number, id: string, key: string, issued_at_ms: number, expires_at_ms: number, } & ({ "type": "set_account_limit", user_id: Id, on: boolean, } | { "type": "temp_ban_ip", ip: string, until_ms: number, } | { "type": "limit_new_conversations", user_id: Id, on: boolean, until_ms: number, } | { "type": "hide_recent_messages", user_id: Id, since_ms: number, on: boolean, } | { "type": "hide_profile", user_id: Id, on: boolean, } | { "type": "delete_user_messages", user_id: Id, on: boolean, } | { "type": "pause_guild_invites", guild_id: Id, on: boolean, until_ms: number, }); +export type Action = { "type": "set_account_limit", user_id: Id, on: boolean, } | { "type": "temp_ban_ip", ip: string, until_ms: number, } | { "type": "limit_new_conversations", user_id: Id, on: boolean, until_ms: number, } | { "type": "hide_recent_messages", user_id: Id, since_ms: number, on: boolean, } | { "type": "hide_profile", user_id: Id, on: boolean, } | { "type": "delete_user_messages", user_id: Id, on: boolean, } | { "type": "pause_guild_invites", guild_id: Id, on: boolean, until_ms: number, }; +export type ActionOutcome = { action_id: string, action_type: string, status: OutcomeStatus, detail: string | null, observed: Observed | null, user_id?: Id, guild_id?: Id, }; export type OutcomeStatus = "applied" | "noop" | "conflict" | "expired" | "ineligible" | "exempt" | "unsupported" | "failed"; export type Observed = { flags: Flags64, deleted: boolean, }; @@ -49,6 +58,7 @@ export const EVENTS_STREAM = 'EVENTS'; export const ACTIONS_STREAM = 'ACTIONS'; export const ACTION_PARTITIONS = 64; export const MAX_ACTION_ATTEMPTS = 10; +export const INVITES_ALREADY_PAUSED = 'already_paused'; export const LISTS_BUCKET = 'lists'; export const SHARED_LISTS = ['ip_blocked', 'email_tld_blocked', 'email_domain_exempt'] as const; export const HEADER = { @@ -80,6 +90,12 @@ export const EVENT_KINDS: ReadonlyArray = [ 'guild_member_banned', 'guild_member_unbanned', 'report_resolved', + 'guild_created', + 'guild_features_changed', + 'invite_created', + 'invite_deleted', + 'guild_member_removed', + 'guild_join_refused', ]; export const EVENT_MAJOR: Record = { registration: 1, @@ -103,6 +119,12 @@ export const EVENT_MAJOR: Record = { guild_member_banned: 1, guild_member_unbanned: 1, report_resolved: 1, + guild_created: 1, + guild_features_changed: 1, + invite_created: 1, + invite_deleted: 1, + guild_member_removed: 1, + guild_join_refused: 1, }; export const EVENT_TTL: Record = { registration: 'never', @@ -126,6 +148,12 @@ export const EVENT_TTL: Record = { guild_member_banned: '3024000', guild_member_unbanned: '3024000', report_resolved: 'never', + guild_created: '3024000', + guild_features_changed: '3024000', + invite_created: '3024000', + invite_deleted: '3024000', + guild_member_removed: '3024000', + guild_join_refused: '3024000', }; export const EVENT_CLASS: Record = { registration: 'fact', @@ -149,10 +177,17 @@ export const EVENT_CLASS: Record = { guild_member_banned: 'fact', guild_member_unbanned: 'fact', report_resolved: 'fact', + guild_created: 'fact', + guild_features_changed: 'fact', + invite_created: 'fact', + invite_deleted: 'fact', + guild_member_removed: 'signal', + guild_join_refused: 'signal', }; export const effectsConsumer = (p: number): string => `effects-${String(p).padStart(2, '0')}`; export function keyToken(key: string): string { if (/^\d+$/.test(key)) return key; return createHash('sha256').update(key).digest('hex').slice(0, 16); } +export const guildKey = (guildId: Id): string => `guild:${guildId}`; export const eventSubject = (kind: EventKind, key: string): string => `evt.in.${kind}.${keyToken(key)}`; diff --git a/fluxer_api/src/api/infrastructure/activity/GuildEvents.ts b/fluxer_api/src/api/infrastructure/activity/GuildEvents.ts new file mode 100644 index 000000000..8b99b2292 --- /dev/null +++ b/fluxer_api/src/api/infrastructure/activity/GuildEvents.ts @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {GuildID, UserID} from '@app/api/BrandedTypes'; +import {emitActivity} from '@app/api/infrastructure/activity/ActivityEvents'; +import { + type FeatureSource, + guildKey, + type JoinSource, + type Meta, + type RemovalKind, +} from '@app/api/infrastructure/activity/Contract.generated'; +import type {Guild} from '@app/api/models/Guild'; +import type {Invite} from '@app/api/models/Invite'; + +export async function emitGuildCreated(guild: Guild): Promise { + await emitActivity( + 'guild_created', + guild.ownerId.toString(), + { + guild_id: guild.id.toString(), + owner_id: guild.ownerId.toString(), + member_count: guild.memberCount, + }, + null, + guild.id.toString(), + ); +} + +export async function emitGuildFeaturesChanged(params: { + previous: Guild; + updated: Guild; + actorId: UserID | null; + source: FeatureSource; + meta?: Meta | null; +}): Promise { + const {previous, updated} = params; + const added = [...updated.features].filter((feature) => !previous.features.has(feature)).sort(); + const removed = [...previous.features].filter((feature) => !updated.features.has(feature)).sort(); + const pausedUntilMs = updated.invitesPausedUntil?.getTime() ?? null; + const pauseChanged = pausedUntilMs !== (previous.invitesPausedUntil?.getTime() ?? null); + if (added.length === 0 && removed.length === 0 && !pauseChanged) return; + const guildId = updated.id.toString(); + await emitActivity( + 'guild_features_changed', + guildKey(guildId), + { + guild_id: guildId, + actor_id: params.actorId?.toString() ?? null, + source: params.source, + added, + removed, + invites_paused_until_ms: pausedUntilMs, + }, + params.meta ?? null, + `${guildId}:${updated.version}`, + ); +} + +export async function emitInviteCreated(invite: Invite): Promise { + if (!invite.guildId) return; + const guildId = invite.guildId.toString(); + await emitActivity( + 'invite_created', + guildKey(guildId), + { + guild_id: guildId, + channel_id: invite.channelId?.toString() ?? null, + code: invite.code, + inviter_id: invite.inviterId?.toString() ?? null, + max_uses: invite.maxUses, + max_age_s: invite.maxAge, + temporary: invite.temporary, + }, + null, + invite.code, + ); +} + +export async function emitInviteDeleted(invite: Invite): Promise { + if (!invite.guildId) return; + const guildId = invite.guildId.toString(); + await emitActivity('invite_deleted', guildKey(guildId), {guild_id: guildId, code: invite.code}, null, invite.code); +} + +export async function emitGuildMemberRemoved(params: { + guildId: GuildID; + userId: UserID; + actorId: UserID | null; + kind: RemovalKind; +}): Promise { + await emitActivity('guild_member_removed', params.userId.toString(), { + guild_id: params.guildId.toString(), + user_id: params.userId.toString(), + actor_id: params.actorId?.toString() ?? null, + kind: params.kind, + }); +} + +export async function emitGuildJoinRefused(params: { + guildId: GuildID; + userId: UserID; + joinSource: JoinSource; +}): Promise { + await emitActivity('guild_join_refused', params.userId.toString(), { + guild_id: params.guildId.toString(), + user_id: params.userId.toString(), + cause: 'invites_paused', + join_source: params.joinSource, + }); +} diff --git a/fluxer_api/src/api/infrastructure/activity/tests/GuildEvents.test.ts b/fluxer_api/src/api/infrastructure/activity/tests/GuildEvents.test.ts new file mode 100644 index 000000000..c1370044c --- /dev/null +++ b/fluxer_api/src/api/infrastructure/activity/tests/GuildEvents.test.ts @@ -0,0 +1,256 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {createTestAccount, setUserACLs} from '@app/api/auth/tests/AuthTestUtils'; +import {createGuildID} from '@app/api/BrandedTypes'; +import {applyPauseGuildInvites, type GuildInvitePauseDeps} from '@app/api/guild/services/GuildInvitePauseApplier'; +import { + acceptInvite, + createChannelInvite, + createGuild, + deleteInvite, + leaveGuild, + setupTestGuildWithMembers, + updateGuild, +} from '@app/api/guild/tests/GuildTestUtils'; +import { + idleActivityEvents, + resetActivityEventsForTests, + startActivityEvents, +} from '@app/api/infrastructure/activity/ActivityEvents'; +import type {ActivityPublisher} from '@app/api/infrastructure/activity/ActivitySpool'; +import type {Event} from '@app/api/infrastructure/activity/Contract.generated'; +import {getGatewayService, getKVClient} from '@app/api/middleware/ServiceRegistry'; +import {getCacheService, getGuildAuditLogService, getGuildRepository} from '@app/api/middleware/ServiceSingletons'; +import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness'; +import {HTTP_STATUS} from '@app/api/test/TestConstants'; +import {createBuilder} from '@app/api/test/TestRequestBuilder'; +import {APIErrorCodes} from '@fluxer/constants/src/ApiErrorCodes'; +import {GuildFeatures} from '@fluxer/constants/src/GuildConstants'; +import type {GuildResponse} from '@fluxer/schema/src/domains/guild/GuildResponseSchemas'; +import {MockKVProvider} from '@pkgs/kv_client/src/testing/MockKVProvider'; +import {afterEach, beforeEach, describe, expect, test} from 'vitest'; + +const HOUR_MS = 3_600_000; + +class CapturingPublisher implements ActivityPublisher { + readonly events: Array = []; + + async publish(_subject: string, payload: string): Promise { + this.events.push(JSON.parse(payload) as Event); + } + + of(kind: K): Array> { + return this.events.filter((event): event is Extract => event.kind === kind); + } +} + +function pauseDeps(): GuildInvitePauseDeps { + return { + guilds: getGuildRepository(), + locks: getCacheService(), + schedule: getKVClient(), + gateway: getGatewayService(), + auditLog: getGuildAuditLogService(), + }; +} + +async function pauseInvites(guildId: string, untilMs: number): Promise { + const outcome = await applyPauseGuildInvites(pauseDeps(), { + v: 1, + id: `a:11:${guildId}:0`, + key: `guild:${guildId}`, + issued_at_ms: Date.now(), + expires_at_ms: Date.now() + 60_000, + type: 'pause_guild_invites', + guild_id: guildId, + on: true, + until_ms: untilMs, + }); + expect(outcome.status).toBe('applied'); +} + +async function storedPause(guildId: string): Promise<{features: Set; until: Date | null}> { + const guild = await getGuildRepository().findUnique(createGuildID(BigInt(guildId))); + return {features: new Set(guild!.features), until: guild!.invitesPausedUntil}; +} + +describe('guild activity events', () => { + let harness: ApiTestHarness; + let publisher: CapturingPublisher; + + beforeEach(async () => { + harness = await createApiTestHarness(); + publisher = new CapturingPublisher(); + await startActivityEvents({publisher, kv: new MockKVProvider()}); + }); + + afterEach(async () => { + resetActivityEventsForTests(); + await harness?.shutdown(); + }); + + test('creating a guild publishes who owns it', async () => { + const owner = await createTestAccount(harness); + const guild = await createGuild(harness, owner.token, 'Fresh guild'); + const [created] = publisher.of('guild_created'); + expect(created?.key).toBe(owner.userId); + expect(created?.data).toEqual({guild_id: guild.id, owner_id: owner.userId, member_count: 1}); + }); + + test('a member feature toggle publishes the actor and the member source', async () => { + const owner = await createTestAccount(harness); + const guild = await createGuild(harness, owner.token, 'Toggle guild'); + await updateGuild(harness, owner.token, guild.id, {features: [...guild.features, GuildFeatures.INVITES_DISABLED]}); + const [changed] = publisher.of('guild_features_changed'); + expect(changed?.key).toBe(`guild:${guild.id}`); + expect(changed?.data).toEqual({ + guild_id: guild.id, + actor_id: owner.userId, + source: 'member', + added: [GuildFeatures.INVITES_DISABLED], + removed: [], + invites_paused_until_ms: null, + }); + }); + + test('creating and deleting an invite publishes both with the invite settings', async () => { + const owner = await createTestAccount(harness); + const guild = await createGuild(harness, owner.token, 'Invite guild'); + const invite = await createChannelInvite(harness, owner.token, guild.system_channel_id!); + await deleteInvite(harness, owner.token, invite.code); + const [created] = publisher.of('invite_created'); + expect(created?.key).toBe(`guild:${guild.id}`); + expect(created?.data).toEqual({ + guild_id: guild.id, + channel_id: guild.system_channel_id, + code: invite.code, + inviter_id: owner.userId, + max_uses: invite.max_uses, + max_age_s: invite.max_age, + temporary: false, + }); + const [deleted] = publisher.of('invite_deleted'); + expect(deleted?.key).toBe(`guild:${guild.id}`); + expect(deleted?.data).toEqual({guild_id: guild.id, code: invite.code}); + }); + + test('leaving and being kicked publish the removal kind and actor', async () => { + const {owner, members, guild} = await setupTestGuildWithMembers(harness, 2); + const [leaver, kicked] = members; + await leaveGuild(harness, leaver!.token, guild.id); + await createBuilder(harness, owner.token) + .delete(`/guilds/${guild.id}/members/${kicked!.userId}`) + .expect(HTTP_STATUS.NO_CONTENT) + .execute(); + await idleActivityEvents(); + expect(publisher.of('guild_member_removed').map((event) => [event.key, event.data])).toEqual([ + [leaver!.userId, {guild_id: guild.id, user_id: leaver!.userId, actor_id: null, kind: 'left'}], + [kicked!.userId, {guild_id: guild.id, user_id: kicked!.userId, actor_id: owner.userId, kind: 'kicked'}], + ]); + }); + + test('a join refused while invites are paused publishes the refusal only for non-members', async () => { + const owner = await createTestAccount(harness); + const joiner = await createTestAccount(harness); + const guild = await createGuild(harness, owner.token, 'Paused guild'); + const invite = await createChannelInvite(harness, owner.token, guild.system_channel_id!); + await pauseInvites(guild.id, Date.now() + HOUR_MS); + for (const account of [owner, joiner]) { + await createBuilder(harness, account.token) + .post(`/invites/${invite.code}`) + .body(null) + .expect(HTTP_STATUS.FORBIDDEN, APIErrorCodes.INVITES_DISABLED) + .execute(); + } + await idleActivityEvents(); + const refusals = publisher.of('guild_join_refused'); + expect(refusals).toHaveLength(1); + const [refused] = refusals; + expect(refused?.key).toBe(joiner.userId); + expect(refused?.data).toEqual({ + guild_id: guild.id, + user_id: joiner.userId, + cause: 'invites_paused', + join_source: 'invite', + }); + }); + + describe('managed invite pause', () => { + let owner: Awaited>; + let guild: GuildResponse; + let untilMs: number; + + beforeEach(async () => { + owner = await createTestAccount(harness); + guild = await createGuild(harness, owner.token, 'Managed pause'); + untilMs = Date.now() + 2 * HOUR_MS; + await pauseInvites(guild.id, untilMs); + }); + + test('the owner turning invites back on ends the pause and drops the notice', async () => { + const paused = await storedPause(guild.id); + expect(paused.until?.getTime()).toBe(untilMs); + expect(paused.features.has(GuildFeatures.RAID_DETECTED)).toBe(true); + const features = [...paused.features].filter((feature) => feature !== GuildFeatures.INVITES_DISABLED); + const updated = await updateGuild(harness, owner.token, guild.id, {features}); + expect(updated.features).not.toContain(GuildFeatures.INVITES_DISABLED); + expect(updated.features).not.toContain(GuildFeatures.RAID_DETECTED); + const after = await storedPause(guild.id); + expect(after.until).toBeNull(); + const memberChange = publisher.of('guild_features_changed').at(-1); + expect(memberChange?.data).toEqual({ + guild_id: guild.id, + actor_id: owner.userId, + source: 'member', + added: [], + removed: [GuildFeatures.INVITES_DISABLED, GuildFeatures.RAID_DETECTED], + invites_paused_until_ms: null, + }); + const joiner = await createTestAccount(harness); + const invite = await createChannelInvite(harness, owner.token, guild.system_channel_id!); + await acceptInvite(harness, joiner.token, invite.code); + }); + + test('an unrelated member edit keeps the pause', async () => { + await updateGuild(harness, owner.token, guild.id, {name: 'Renamed'}); + expect((await storedPause(guild.id)).until?.getTime()).toBe(untilMs); + }); + + test('a staff edit of the pause features takes the pause over from the timer', async () => { + const admin = await setUserACLs(harness, await createTestAccount(harness), [ + 'admin:authenticate', + 'guild:update:features', + ]); + await createBuilder(harness, admin.token) + .patch(`/admin/guilds/${guild.id}`) + .body({remove_features: [GuildFeatures.RAID_DETECTED]}) + .expect(HTTP_STATUS.OK) + .execute(); + const after = await storedPause(guild.id); + expect(after.until).toBeNull(); + expect(after.features.has(GuildFeatures.INVITES_DISABLED)).toBe(true); + expect(after.features.has(GuildFeatures.RAID_DETECTED)).toBe(false); + expect(publisher.of('guild_features_changed').at(-1)?.data).toEqual({ + guild_id: guild.id, + actor_id: admin.userId, + source: 'staff', + added: [], + removed: [GuildFeatures.RAID_DETECTED], + invites_paused_until_ms: null, + }); + }); + + test('a staff edit of other features leaves the timer running', async () => { + const admin = await setUserACLs(harness, await createTestAccount(harness), [ + 'admin:authenticate', + 'guild:update:features', + ]); + await createBuilder(harness, admin.token) + .patch(`/admin/guilds/${guild.id}`) + .body({add_features: [GuildFeatures.VERIFIED]}) + .expect(HTTP_STATUS.OK) + .execute(); + expect((await storedPause(guild.id)).until?.getTime()).toBe(untilMs); + }); + }); +}); diff --git a/fluxer_api/src/api/invite/InviteService.ts b/fluxer_api/src/api/invite/InviteService.ts index b08fc1714..edde0a807 100644 --- a/fluxer_api/src/api/invite/InviteService.ts +++ b/fluxer_api/src/api/invite/InviteService.ts @@ -13,6 +13,7 @@ import type {ChannelService} from '@app/api/channel/services/ChannelService'; import {THREAD_FEATURE_CHANNEL_TYPES, type ThreadViewer} from '@app/api/experiment/ChannelThreadsGate'; import type {GuildAuditLogService} from '@app/api/guild/GuildAuditLogService'; import type {GuildService} from '@app/api/guild/services/GuildService'; +import {emitGuildJoinRefused, emitInviteCreated, emitInviteDeleted} from '@app/api/infrastructure/activity/GuildEvents'; import type {InviteRepository} from '@app/api/invite/InviteRepository'; import {Logger} from '@app/api/Logger'; import type {LimitConfigService} from '@app/api/limits/LimitConfigService'; @@ -243,6 +244,7 @@ export class InviteService { action: 'create', auditLogReason, }); + await emitInviteCreated(newInvite); } return {invite: newInvite, isNew: true}; } @@ -291,13 +293,18 @@ export class InviteService { if ((guild.disabledOperations & GuildOperations.INSTANT_INVITES) !== 0) { throw new FeatureTemporarilyDisabledError(); } - if (guild.features.has(GuildFeatures.INVITES_DISABLED)) { - throw new InvitesDisabledError(); - } + const vanityCode = guild.vanityUrlCode ? vanityCodeToInviteCode(guild.vanityUrlCode) : null; + const isVanityInvite = invite.code === vanityCode; const existingMember = await this.apiContext.services.gateway.hasGuildMember({ guildId: invite.guildId, userId, }); + if (guild.features.has(GuildFeatures.INVITES_DISABLED)) { + if (!existingMember) { + await emitGuildJoinRefused({guildId: guild.id, userId, joinSource: isVanityInvite ? 'vanity' : 'invite'}); + } + throw new InvitesDisabledError(); + } if (existingMember) { return invite; } @@ -308,8 +315,6 @@ export class InviteService { throw new TemporaryInviteRequiresPresenceError(); } } - const vanityCode = guild.vanityUrlCode ? vanityCodeToInviteCode(guild.vanityUrlCode) : null; - const isVanityInvite = invite.code === vanityCode; const guildId = invite.guildId; const reservedInvite = await this.reserveInviteUse(invite); await this.withReservedInviteUse(reservedInvite, () => @@ -459,6 +464,7 @@ export class InviteService { action: 'delete', auditLogReason, }); + await emitInviteDeleted(invite); } async dispatchInviteCreate( diff --git a/fluxer_api/src/api/models/Guild.ts b/fluxer_api/src/api/models/Guild.ts index 25ba250ce..a61d0e0ee 100644 --- a/fluxer_api/src/api/models/Guild.ts +++ b/fluxer_api/src/api/models/Guild.ts @@ -49,6 +49,7 @@ export class Guild { readonly auditLogsIndexedAt: Date | null; readonly membersIndexedAt: Date | null; readonly messageHistoryCutoff: Date | null; + readonly invitesPausedUntil: Date | null; readonly version: number; constructor(row: GuildRow) { @@ -89,6 +90,7 @@ export class Guild { this.auditLogsIndexedAt = row.audit_logs_indexed_at ?? null; this.membersIndexedAt = row.members_indexed_at ?? null; this.messageHistoryCutoff = row.message_history_cutoff ?? null; + this.invitesPausedUntil = row.invites_paused_until ?? null; this.version = row.version; } @@ -128,6 +130,7 @@ export class Guild { audit_logs_indexed_at: this.auditLogsIndexedAt, members_indexed_at: this.membersIndexedAt, message_history_cutoff: this.messageHistoryCutoff, + invites_paused_until: this.invitesPausedUntil, version: this.version, }; } diff --git a/fluxer_api/src/api/search/guild/tests/GuildSearchSerializer.test.ts b/fluxer_api/src/api/search/guild/tests/GuildSearchSerializer.test.ts index 44ff70ba2..81e21d3c1 100644 --- a/fluxer_api/src/api/search/guild/tests/GuildSearchSerializer.test.ts +++ b/fluxer_api/src/api/search/guild/tests/GuildSearchSerializer.test.ts @@ -43,6 +43,7 @@ function guildRow(features: Set): GuildRow { audit_logs_indexed_at: null, members_indexed_at: null, message_history_cutoff: null, + invites_paused_until: null, version: 1, }; } diff --git a/fluxer_api/src/api/user/services/AccountStateApplier.ts b/fluxer_api/src/api/user/services/AccountStateApplier.ts index 498a7369b..13a371f8a 100644 --- a/fluxer_api/src/api/user/services/AccountStateApplier.ts +++ b/fluxer_api/src/api/user/services/AccountStateApplier.ts @@ -123,7 +123,7 @@ function observedOf(user: User): Observed { } export function outcomeOf( - env: Pick & {type: string; user_id?: string}, + env: Pick & {type: string; user_id?: string; guild_id?: string}, status: OutcomeStatus, user: User | null = null, detail: string | null = null, @@ -136,6 +136,7 @@ export function outcomeOf( observed: user ? observedOf(user) : null, }; if (env.user_id) outcome.user_id = env.user_id; + if (env.guild_id) outcome.guild_id = env.guild_id; return outcome; } diff --git a/fluxer_api/src/api/utils/tests/MutualGuildDmAccess.test.ts b/fluxer_api/src/api/utils/tests/MutualGuildDmAccess.test.ts index 591cb2fed..b71fe9511 100644 --- a/fluxer_api/src/api/utils/tests/MutualGuildDmAccess.test.ts +++ b/fluxer_api/src/api/utils/tests/MutualGuildDmAccess.test.ts @@ -47,6 +47,7 @@ function createGuild(overrides: Partial = {}): Guild { audit_logs_indexed_at: null, members_indexed_at: null, message_history_cutoff: null, + invites_paused_until: null, version: 1, ...overrides, }); diff --git a/fluxer_api/src/api/worker/AccountActionConsumer.ts b/fluxer_api/src/api/worker/AccountActionConsumer.ts index b892a38c5..a5880a620 100644 --- a/fluxer_api/src/api/worker/AccountActionConsumer.ts +++ b/fluxer_api/src/api/worker/AccountActionConsumer.ts @@ -1,5 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +import {applyPauseGuildInvites, type GuildInvitePauseDeps} from '@app/api/guild/services/GuildInvitePauseApplier'; import {emitActivity} from '@app/api/infrastructure/activity/ActivityEvents'; import {workerMeta} from '@app/api/infrastructure/activity/ActivityMeta'; import { @@ -27,6 +28,7 @@ import {AckPolicy, type ConsumerMessages, type JetStreamClient, type JsMsg} from export interface AccountActionDeps { js: Pick; state: AccountStateDeps; + guildInvites: GuildInvitePauseDeps; publishOutcome?: (key: string, outcome: ActionOutcome) => Promise; now?: () => number; retryDelayMs?: number; @@ -79,6 +81,8 @@ export function applyAction(deps: AccountActionDeps, env: ActionEnvelope): Promi return applyHideRecentMessages(deps.state, env); case 'delete_user_messages': return applyDeleteUserMessages(deps.state, env); + case 'pause_guild_invites': + return applyPauseGuildInvites(deps.guildInvites, env); default: return Promise.resolve(outcomeOf(env as ActionEnvelope, 'unsupported')); } diff --git a/fluxer_api/src/api/worker/WorkerDependencies.ts b/fluxer_api/src/api/worker/WorkerDependencies.ts index 497ceac03..65382aa04 100644 --- a/fluxer_api/src/api/worker/WorkerDependencies.ts +++ b/fluxer_api/src/api/worker/WorkerDependencies.ts @@ -11,6 +11,7 @@ import {DonationRepository} from '@app/api/donation/DonationRepository'; import type {FavoriteMemeRepository} from '@app/api/favorite_meme/FavoriteMemeRepository'; import type {GuildAuditLogService} from '@app/api/guild/GuildAuditLogService'; import type {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; +import type {GuildInvitePauseDeps} from '@app/api/guild/services/GuildInvitePauseApplier'; import type {GuildService} from '@app/api/guild/services/GuildService'; import type {AvatarService} from '@app/api/infrastructure/AvatarService'; import type {IPurgeQueue} from '@app/api/infrastructure/CachePurgeQueue'; @@ -281,3 +282,13 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS }, }; } + +export function guildInvitePauseDeps(deps: WorkerDependencies): GuildInvitePauseDeps { + return { + guilds: deps.guildRepository, + locks: deps.cacheService, + schedule: deps.kvClient, + gateway: deps.gatewayService, + auditLog: deps.guildAuditLogService, + }; +} diff --git a/fluxer_api/src/api/worker/WorkerLaneConfig.ts b/fluxer_api/src/api/worker/WorkerLaneConfig.ts index 7830e3b8d..10aa2456a 100644 --- a/fluxer_api/src/api/worker/WorkerLaneConfig.ts +++ b/fluxer_api/src/api/worker/WorkerLaneConfig.ts @@ -59,6 +59,7 @@ const LANE_CONFIG = { 'bulkBanFileShas', 'bulkDeleteMessagesForUsers', 'drainActivitySpool', + 'liftExpiredGuildInvitePauses', ] as const, retiredTasks: ['sendScheduledMessage', 'finalizeNcmecAttachmentReport'], concurrency: 8, diff --git a/fluxer_api/src/api/worker/WorkerMain.ts b/fluxer_api/src/api/worker/WorkerMain.ts index a117424ee..3e8abec0e 100644 --- a/fluxer_api/src/api/worker/WorkerMain.ts +++ b/fluxer_api/src/api/worker/WorkerMain.ts @@ -40,7 +40,11 @@ import {queueBlocklistFeedStartupJobs} from '@app/api/worker/BlocklistFeedStartu import {CronScheduler} from '@app/api/worker/CronScheduler'; import {JetStreamWorkerQueue, JOBS_STREAM_MAX_AGE_MS} from '@app/api/worker/JetStreamWorkerQueue'; import {clearWorkerDependencies, setWorkerDependencies} from '@app/api/worker/WorkerContext'; -import {initializeWorkerDependencies, type WorkerDependencies} from '@app/api/worker/WorkerDependencies'; +import { + guildInvitePauseDeps, + initializeWorkerDependencies, + type WorkerDependencies, +} from '@app/api/worker/WorkerDependencies'; import {WorkerHeartbeat} from '@app/api/worker/WorkerHeartbeat'; import { resolveCronSchedulerEnabled, @@ -108,6 +112,7 @@ function registerCronJobs(cron: CronScheduler, jobsStreamMaxAgeMs: number): void } cron.upsert('flushUserActivityBuffer', 'flushUserActivityBuffer', {}, '*/10 * * * * *', {ledger: false}); cron.upsert('drainActivitySpool', 'drainActivitySpool', {}, '*/5 * * * * *', {ledger: false}); + cron.upsert('liftExpiredGuildInvitePauses', 'liftExpiredGuildInvitePauses', {}, '0 * * * * *', {ledger: false}); cron.upsert('archiveInactiveThreads', 'archiveInactiveThreads', {}, '0 * * * * *', { ledger: false, enabled: () => true, @@ -328,6 +333,7 @@ export async function startWorkerMain(): Promise { }, dependencies.channelRepository, ), + guildInvites: guildInvitePauseDeps(dependencies), }); Logger.info('Account action consumer started'); } diff --git a/fluxer_api/src/api/worker/WorkerTaskRegistry.ts b/fluxer_api/src/api/worker/WorkerTaskRegistry.ts index 4e9869496..3525e6d73 100644 --- a/fluxer_api/src/api/worker/WorkerTaskRegistry.ts +++ b/fluxer_api/src/api/worker/WorkerTaskRegistry.ts @@ -27,6 +27,7 @@ import harvestGuildData from '@app/api/worker/tasks/HarvestGuildData'; import harvestUserData from '@app/api/worker/tasks/HarvestUserData'; import indexChannelMessages from '@app/api/worker/tasks/IndexChannelMessages'; import indexGuildMembers from '@app/api/worker/tasks/IndexGuildMembers'; +import liftExpiredGuildInvitePauses from '@app/api/worker/tasks/LiftExpiredGuildInvitePauses'; import messageShred from '@app/api/worker/tasks/MessageShred'; import pollAppStoreNotificationHistory from '@app/api/worker/tasks/PollAppStoreNotificationHistory'; import pollGooglePlayVoidedPurchases from '@app/api/worker/tasks/PollGooglePlayVoidedPurchases'; @@ -99,6 +100,7 @@ export const workerTasks: Record = { harvestUserData, indexChannelMessages, indexGuildMembers, + liftExpiredGuildInvitePauses, messageShred, processAssetDeletionQueue, processCachePurgeQueue, diff --git a/fluxer_api/src/api/worker/tasks/LiftExpiredGuildInvitePauses.ts b/fluxer_api/src/api/worker/tasks/LiftExpiredGuildInvitePauses.ts new file mode 100644 index 000000000..214eed4de --- /dev/null +++ b/fluxer_api/src/api/worker/tasks/LiftExpiredGuildInvitePauses.ts @@ -0,0 +1,15 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {liftExpiredGuildInvitePauses} from '@app/api/guild/services/GuildInvitePauseApplier'; +import {getWorkerDependencies} from '@app/api/worker/WorkerContext'; +import {guildInvitePauseDeps} from '@app/api/worker/WorkerDependencies'; +import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask'; + +const BATCH_SIZE = 100; + +const liftExpiredGuildInvitePausesTask: WorkerTaskHandler = async (_payload, helpers) => { + const lifted = await liftExpiredGuildInvitePauses(guildInvitePauseDeps(getWorkerDependencies()), BATCH_SIZE); + if (lifted > 0) helpers.logger.info({lifted}, 'Lifted expired guild invite pauses'); +}; + +export default liftExpiredGuildInvitePausesTask; diff --git a/fluxer_api/src/api/worker/tests/AccountActionConsumer.test.ts b/fluxer_api/src/api/worker/tests/AccountActionConsumer.test.ts index 361daa885..acc49ae5c 100644 --- a/fluxer_api/src/api/worker/tests/AccountActionConsumer.test.ts +++ b/fluxer_api/src/api/worker/tests/AccountActionConsumer.test.ts @@ -208,7 +208,12 @@ function harness(): Harness { authored: authored as unknown as AccountStateDeps['authored'], now: () => NOW, }; - h.deps = {js: {} as AccountActionDeps['js'], state, now: () => NOW}; + h.deps = { + js: {} as AccountActionDeps['js'], + state, + guildInvites: {} as AccountActionDeps['guildInvites'], + now: () => NOW, + }; return h; } diff --git a/tools/dev/cassandra_target_schema.json b/tools/dev/cassandra_target_schema.json index f8e460263..31bc00fbb 100644 --- a/tools/dev/cassandra_target_schema.json +++ b/tools/dev/cassandra_target_schema.json @@ -5903,6 +5903,10 @@ { "name": "content_warning_text", "type": "text" + }, + { + "name": "invites_paused_until", + "type": "timestamp" } ], "primary_key": "((guild_id))",