# 运营中心 - 数据模型文档（v1：M365 休眠账号）

> **版本**: v0.2
> **最后更新**: 2026-04-30
> **维护者**: 后端团队

---

## ✅ 机器读取区（必填）

### Schema 摘要

| 字段 | 内容 |
|------|------|
| Schema 文件 | `backend/prisma/schema/platform_ops_center.prisma` |
| 业务域 | 运营中心 / M365 休眠账号 |
| 核心实体 | `M365User`（当前态）、`M365UserActivityChange`（变更事件）；同步元数据**复用** `platform_automation` 的 `AutomationTask` + `AutomationExecution`，不自建 SyncBatch 表 |
| 同步基础设施 | 复用 `backend/prisma/schema/platform_automation.prisma` 的 `AutomationTask` + `AutomationExecution`，与 LDAP_SYNC / DINGTALK_SYNC / ADP_SYNC 等已有同步模块共享调度、统计、监控、告警基础设施。仅需在 `AutomationTaskType` enum 加一个值 `M365_DORMANT_SYNC` |
| 命名约定 | Prisma model 用 PascalCase，PostgreSQL 表名用 `@@map` 显式声明（snake_case） |
| 时间字段 | 所有 `DateTime` 字段使用 `@db.Timestamptz(6)`（PostgreSQL `timestamp with time zone`），全部按 UTC 存储与计算 |
| UPN 大小写 | `userPrincipalName` 字段使用 PostgreSQL `citext`（case-insensitive text）类型存储；JOIN / 比较前**统一 `toLowerCase()`** 再做（Graph `/users` 与活动报告 CSV 对 UPN 大小写不保证一致，是已知坑） |
| 主键策略 | `graphUserId`（Graph objectId）是稳定唯一标识，业务主键以 `(organizationId, graphUserId)` 为准；`userPrincipalName` 仅作展示与 JOIN 报告的临时键，**不可作为业务主键**——deleted user 的 UPN 可能被复用给新用户 |

### 设计模式

**三层架构**：

| 层 | 数据 | 表 | 来源 |
|---|---|---|---|
| L1 同步编排元数据 | 谁触发、何时、status、duration、错误、统计、直方图 | `AutomationExecution` | **复用 platform_automation** |
| L2 用户当前态 | 6 个时间字段、licenses、accountEnabled、daysInactive | `M365User` | **本模块自建** |
| L3 用户活动历史 | Alice 何时从活跃变休眠、license 何时新增/移除 | `M365UserActivityChange` | **本模块自建** |

`M365User` 只存"最新一次同步看到的状态"，每次同步 upsert（`graphUserId` 唯一）。
`M365UserActivityChange` 仅当某字段相比上次同步**有变化**时追加一行。
两表通过 FK 指向 `AutomationExecution.id`，跨 schema 关联。

### L1 — 复用 `AutomationTask` + `AutomationExecution`

**不自建 SyncBatch 表**。理由 + 现有基础设施引用见 `[LRN-20260425-001]`、参考 `dingtalk-scheduler.service.ts` / `adp-scheduler.service.ts`。

#### 需要的最小改动

`platform_automation.prisma` 的 `AutomationTaskType` enum 加一个值：

```prisma
enum AutomationTaskType {
  LDAP_SYNC
  DINGTALK_SYNC
  ADP_SYNC
  M365_DORMANT_SYNC   // 新增
  // ...
}
```

#### 字段映射（M365 概念 → AutomationExecution 字段）

| 业务字段 | 落在哪 |
|---|---|
| 触发人 | `triggeredBy` |
| 触发方式（manual/scheduled） | `triggerType` |
| 状态 | `status`（`PENDING` / `RUNNING` / `SUCCESS` / `FAILED` / `TIMEOUT` / `CANCELLED`） |
| 起止时间 | `startedAt` / `completedAt` / `duration` |
| 错误码 + 摘要 | `error` 字符串，约定格式 `"{ERROR_CODE}: {message}"`（如 `"AZURE_CREDENTIAL_MISSING: ..."`），结构化解析交给前端 |
| 详细日志 | `logs` |
| 业务统计 + 直方图 | `result: Json`（结构见下） |

#### `AutomationExecution.result` JSON 结构（M365_DORMANT_SYNC 专用）

```json
{
  "totalUsers": 4321,
  "changedUsers": 87,
  "inactiveDistribution": {
    "never": 21,
    "buckets": [
      { "from": 0,   "to": 30,  "count": 3200 },
      { "from": 30,  "to": 60,  "count": 200 },
      { "from": 60,  "to": 90,  "count": 100 },
      { "from": 90,  "to": 180, "count": 150 },
      { "from": 180, "to": 365, "count": 200 },
      { "from": 365, "to": null,"count": 450 }
    ]
  }
}
```

- `never`：`lastAnyActivity` 为 null 的用户数。
- 桶左闭右开 `[from, to)`；最后一桶 `to=null` 表示 `>= from`。
- v1 固定六桶（30/60/90/180/365 + 365+）；后续改桶定义不影响旧 execution 的渲染。
- success 时必填，failed 时省略。

#### 并发互斥

**不要自建 partial unique index**——查现有同步模块的并发控制实现（`dingtalk-scheduler.service.ts` 等），按惯例在 service 层做：

```ts
const running = await prisma.automationExecution.findFirst({
  where: { task: { type: 'M365_DORMANT_SYNC' }, status: 'RUNNING' }
});
if (running) throw new ConflictException('SYNC_IN_PROGRESS');
```

如发现项目尚无统一惯例（多个模块各做各的），可以在本模块用 service-level mutex，**仍不要改 platform_automation schema**——那会影响所有同步模块。

#### 多租户隔离的注意点

`AutomationTask` / `AutomationExecution` **没有 `organizationId`**——v1 单租户假设下没问题（PRD §"不做的"已明示）。多租户场景下可走的路径：

1. 每个 organization 跑独立的 `AutomationTask`（`code` 字段含 orgId 区分）
2. v1.x 评估给 platform_automation 加 organizationId（项目级改动，不在本模块作用域）

`M365User.organizationId` 仍是 DataScope 标准字段，列表查询走它，不依赖 AutomationExecution 的隔离。

---

#### `M365User`（用户当前态，每次同步 upsert）

| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| id | UUID | ✅ | 主键 |
| organizationId | UUID | ✅ | 标准字段 |
| createdById | UUID | ✅ | 首次同步触发人（追溯用） |
| createdAt | DateTime | ✅ | 首次发现该用户的时间 |
| updatedAt | DateTime | ✅ | 最近一次同步看到该用户的时间 |
| graphUserId | String | ✅ | Graph 返回的 objectId |
| userPrincipalName | citext | ✅ | UPN，case-insensitive，JOIN 前统一 lowercase |
| displayName | String | ❌ | |
| mail | String | ❌ | |
| department | String | ❌ | |
| jobTitle | String | ❌ | |
| accountEnabled | Boolean | ✅ | |
| hasLicense | Boolean | ✅ | 派生字段：`licenses.length > 0`，冗余存储以便走索引筛选 |
| licenses | Json | ✅ | License 详情数组：`[{ skuId, skuPartNumber, displayName }]`；空数组表示无 license |
| lastSignInDateTime | DateTime | ❌ | 6 个时间字段，永远是最新值 |
| lastNonInteractiveSignInDateTime | DateTime | ❌ | |
| lastEmailActivity | DateTime | ❌ | |
| lastOneDriveActivity | DateTime | ❌ | |
| lastTeamsActivity | DateTime | ❌ | |
| lastSharePointActivity | DateTime | ❌ | |
| lastAnyActivity | DateTime | ❌ | 6 字段 max；全 null 时为 null |
| daysInactive | Int | ❌ | 基于"最后一次 SUCCESS execution 的 startedAt"计算 |
| lastSeenInExecutionId | UUID | ✅ | 最近一次见到该用户的 execution（**跨 schema FK** → `platform_automation.AutomationExecution.id`） |
| firstSeenInExecutionId | UUID | ✅ | 首次见到该用户的 execution（同上） |

唯一约束：`(organizationId, graphUserId)`。

索引：
- `(organizationId, daysInactive desc)` — 列表默认排序
- `(organizationId, userPrincipalName)` — 关键字搜索
- `(organizationId, lastSeenInExecutionId)` — 找上次同步缺失的用户（已离职 / 已禁用消失等场景，v1.1 用）

---

#### `M365UserActivityChange`（变更事件流，只追加）

| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| id | UUID | ✅ | 主键 |
| organizationId | UUID | ✅ | 标准字段 |
| createdAt | DateTime | ✅ | = execution.startedAt（写入时刻） |
| userId | UUID | ✅ | FK → M365User.id (ON DELETE CASCADE) |
| executionId | UUID | ✅ | **跨 schema FK** → `platform_automation.AutomationExecution.id` |
| field | Enum | ✅ | `lastSignInDateTime` / `lastNonInteractiveSignInDateTime` / `lastEmailActivity` / `lastOneDriveActivity` / `lastTeamsActivity` / `lastSharePointActivity` / `accountEnabled` / `hasLicense` / `licenses` |
| previousValue | String | ❌ | 上次的值（DateTime / Boolean / JSON 数组字符串） |
| currentValue | String | ❌ | 本次的值 |

`licenses` 字段变更时，`previousValue` / `currentValue` 存 JSON 数组字符串（如 `'["ENTERPRISEPACK","EMS"]'` 仅含 skuPartNumber 列表，简化 diff 比较）。

**只在字段值真的变了时才写**（`previousValue ≠ currentValue`）。同一次同步同一用户多个字段变化 → 多行。

索引：
- `(userId, createdAt desc)` — 用户时间线主查询
- `(organizationId, executionId)` — 按 execution 查 change（趋势 / 审计）
- `(organizationId, field, createdAt desc)` — 按字段类型聚合（如"过去 30 天有多少人重新开始登录"）

不设 `createdById`：change 是系统观察记录而非用户行为，触发人通过 `executionId → AutomationExecution.triggeredBy` 追溯。

---

## 🧭 人类阅读区

### 设计取舍

#### 为什么不再按批次落全量快照？

旧设计（v0.1）每次同步 = N 个用户的全量快照。代价：

- 5000 用户 × 周同步 × 1 年 ≈ **26 万行**，其中 90% 是无变化的重复数据（休眠用户的活动时间字段就是不变）。
- 查"Alice 的活动时间线"要扫所有批次拼出来。
- 趋势查询要跨批次聚合 N×M 行。

新设计：

- `M365User` 只存 N 行（5000 行常驻），列表查询零成本。
- `M365UserActivityChange` 真正变化才写——估算每周新增 100~500 行（活跃用户的活动时间在动），1 年 ≈ 5k~25k 行，比旧方案省 90%+。
- 趋势从 `AutomationExecution.result.inactiveDistribution` 直接读，单次同步 1 行，零扫描。

**信息无损**：当前态 + 全量变更事件 = 可以重放出任意时间点的快照（按需，不预先物化）。

#### 为什么趋势用直方图分桶而不是固定阈值的预聚合？

让前端能"拖 slider 看不同阈值下的休眠数"。固定 90/180/365 三档预聚合不灵活；六桶分布数据让前端任意阈值都能近似算（落在桶中间的阈值取下一桶起累加，或线性插值），单次同步 ≈ 200 字节 JSON，可忽略。

桶不够细（如 30 天粒度看不出 7 天周期）的话，后续改桶定义即可——不影响旧批次（旧批次按旧桶渲染），不需要数据迁移。

#### 为什么 `daysInactive` 写在 `M365User` 上？

写入时基于"本次 batch 的 startedAt"算一次，避免每次列表查询都 `GREATEST(...)` + 算差值，让 `WHERE days_inactive >= ?` 可走索引。代价：列表查询的 `daysInactive` 是"上次同步那一刻的天数"，不是"实时天数"——业务可接受（毕竟同步本身就是离散的时点）。前端展示加上"截至 {batch.startedAt} 的休眠天数"标签消除歧义。

#### 为什么不删除消失的用户？

Graph 返回的用户列表可能因离职 / 禁用 / 删除而缩水。v1 不在 `M365User` 上做删除——通过 `lastSeenInBatchId < latestBatchId` 判定"上次同步未见到"，前端在用户详情里给个标签（"上次同步未在 Graph 中见到，可能已禁用"）。v1.1 再决定是否做硬删除/归档。

### 标准字段合规性

本模块自建的两张表都带：`id` / `organizationId` / `createdAt` / `updatedAt`：
- `M365User.createdById` ✅（首次发现的同步触发人）
- `M365UserActivityChange.createdById` ❌（系统观察，不归属用户行为）——属于"事件流型表"的合理偏离，通过 `executionId → AutomationExecution.triggeredBy` 追溯触发人。

`AutomationTask` / `AutomationExecution` 是项目共享表，不在本模块作用域；它们没有 `organizationId`（v1 单租户假设下接受，详见 L1 段"多租户隔离的注意点"）。

`departmentId` / `regionId` 不适用——同步动作是组织级别。

### 迁移策略

- 新建 `backend/prisma/schema/platform_ops_center.prisma`（多文件 schema）。
- 改 `platform_automation.prisma` 的 `AutomationTaskType` enum，加 `M365_DORMANT_SYNC` 一个值（**单独迁移文件，与本模块两表分开**？还是合一？看项目惯例——`platform_automation` enum 改动跨域，建议合在同一个迁移文件以保证 atomic）。
- 单一迁移文件包含：enum 加值 + 两张表（M365User / M365UserActivityChange）+ 索引 + 跨 schema FK。
- 不涉及历史数据回填。

#### Prisma 不支持的部分需手动编辑生成的 migration.sql

执行 `prisma migrate dev --create-only` 生成 SQL 后**手动编辑**插入 raw SQL，**注意顺序**：

**`CREATE EXTENSION` 必须放在文件最顶部**（在所有 `CREATE TABLE` 之前），否则 CREATE TABLE 引用 citext 类型时报错：
```sql
CREATE EXTENSION IF NOT EXISTS citext;
```
Prisma schema 里字段类型用 `String @db.Citext`。

最终 migration.sql 结构：

```sql
-- 1. 扩展
CREATE EXTENSION IF NOT EXISTS citext;

-- 2. ALTER TYPE "platform_automation"."AutomationTaskType" ADD VALUE 'M365_DORMANT_SYNC'
-- 3. Prisma 自动生成的 CREATE TABLE / CREATE INDEX / FK
-- (不要改这部分)
```

不再需要 partial unique index（并发互斥改在 service 层做，见 L1 段）。

#### Foreign Key 删除策略

| FK | ON DELETE |
|---|---|
| `M365UserActivityChange.userId → M365User` | `CASCADE`（用户删除时事件流跟着删） |
| `M365UserActivityChange.executionId → AutomationExecution` | `RESTRICT`（不允许删 execution，否则审计断链）|
| `M365User.firstSeenInExecutionId / lastSeenInExecutionId → AutomationExecution` | `RESTRICT`（同上） |

> ⚠️ 跨 schema FK 在 Prisma 多文件 schema 下需要用 `String @db.Uuid` + 手动维护引用完整性，**或**确认项目 PostgreSQL 是否启用了跨 schema FK 支持。参考已有跨 schema FK 实现（grep `@db.Uuid` + 跨域 model）。

### organizationId 来源

所有查询 / 写入的 `organizationId` 都来自 JWT payload，由项目 `DataScopeMiddleware` 自动注入到 controller / service 上下文（项目惯例，与其他模块一致）。v1 单租户假设下都是同一个值，但代码必须按多租户模式写——即所有 `findMany` / `create` 都必须带 `organizationId` filter，不能假设全表只有一份数据。

### 同步算法（写入流程伪代码）

```
on POST /sync:
  // 1. 取本租户的 AutomationTask（启动时已 upsert 一条 type=M365_DORMANT_SYNC 的任务）
  task = await prisma.automationTask.findFirstOrThrow({ where: { type: 'M365_DORMANT_SYNC' } })

  // 2. 并发互斥（service-level）
  const running = await prisma.automationExecution.findFirst({ where: { taskId: task.id, status: 'RUNNING' } })
  if (running) throw ConflictException('SYNC_IN_PROGRESS')

  // 3. 创建 execution
  execution = await prisma.automationExecution.create({
    data: { taskId: task.id, status: 'RUNNING', triggerType: 'manual', triggeredBy: actorId, startedAt: now() }
  })

  try:
    // 1. 拉 license SKU 字典（本租户已订阅的 SKU 列表）
    //    注意：Graph SubscribedSku 对象只有 skuId + skuPartNumber，没有用户友好 displayName，
    //    必须配静态映射表（见下方 SKU_DISPLAY_NAMES），缺失时降级用 skuPartNumber 兜底。
    subscribedSkus = await graph.GET('/subscribedSkus')  // [{skuId, skuPartNumber, servicePlans}]
    skuMap = subscribedSkus.reduce((acc, s) => ({
      ...acc,
      [s.skuId]: {
        skuId: s.skuId,
        skuPartNumber: s.skuPartNumber,
        displayName: SKU_DISPLAY_NAMES[s.skuPartNumber] ?? s.skuPartNumber  // 兜底
      }
    }), {})

    // 2. 拉所有用户（分页）+ signInActivity + assignedLicenses
    //    signInActivity 必须显式 $select 才返；$top 最大 999；超过要 follow @odata.nextLink
    SELECT_FIELDS = 'id,userPrincipalName,displayName,mail,department,jobTitle,accountEnabled,signInActivity,assignedLicenses'
    graphUsers = []
    nextUrl = `/users?$select=${SELECT_FIELDS}&$top=999`
    while nextUrl:
      page = await graph.GET(nextUrl)
      graphUsers.push(...page.value)
      nextUrl = page['@odata.nextLink'] ?? null

    // 3. 拉 4 份活动报告（返 text/csv，UTF-8 with BOM），用 csv-parse/sync 解析
    //    UPN 在 CSV 里和 /users 返回的 case 不保证一致，统一 lowercase 后入 map
    activityMaps = {}
    for report in ['EmailActivity', 'OneDriveActivity', 'TeamsUserActivity', 'SharePointActivity']:
      csvText = await graph.GET(`/reports/get${report}UserDetail(period='D180')`)
      // 检测脱敏：如果首行 UPN 是 hash 形式（无 @ 符号或全 hex），抛 REPORTS_OBFUSCATED
      rows = parse(csvText, { columns: true, skip_empty_lines: true })  // csv-parse/sync
      if (rows[0] && !rows[0]['User Principal Name'].includes('@')):
        throw new Error('REPORTS_OBFUSCATED')
      activityMaps[report] = rows.reduce((acc, row) => ({
        ...acc,
        [row['User Principal Name'].toLowerCase()]: parseDate(row['Last Activity Date'])
      }), {})

    // 4. 合并 + diff + 批量写入（每 200 用户一批 commit，避免长事务 + 减少 connection 占用）
    for u in graphUsers:
      upnLower = u.userPrincipalName.toLowerCase()
      enriched = {
        ...u,
        userPrincipalName: u.userPrincipalName,  // 原 case 入库（citext 列存储不区分大小写）
        licenses: (u.assignedLicenses ?? []).map(a => skuMap[a.skuId]).filter(Boolean),
        hasLicense: (u.assignedLicenses ?? []).length > 0,
        lastEmailActivity: activityMaps.EmailActivity[upnLower],
        lastOneDriveActivity: activityMaps.OneDriveActivity[upnLower],
        lastTeamsActivity: activityMaps.TeamsUserActivity[upnLower],
        lastSharePointActivity: activityMaps.SharePointActivity[upnLower],
      }
      enriched.lastAnyActivity = max(6 个时间字段)
      enriched.daysInactive = lastAnyActivity ? floor((execution.startedAt - lastAnyActivity) / 1day) : null

      existing = M365User.findUnique(organizationId, graphUserId=u.id)
      if existing:
        diffs = computeDiffs(existing, enriched)  // 6 时间字段 + accountEnabled + hasLicense + licenses
        if diffs.length > 0:
          insertMany(M365UserActivityChange, diffs.map(d => ({...d, executionId: execution.id, userId: existing.id})))
        update M365User set {...enriched, lastSeenInExecutionId=execution.id, updatedAt=now}
      else:
        create M365User({...enriched, firstSeenInExecutionId=execution.id, lastSeenInExecutionId=execution.id, createdById=actorId})
        // 新用户不写 change 事件（没有"之前的状态"）

    distribution = computeHistogram(M365User where organizationId=...)
    update execution set {
      status: 'SUCCESS',
      completedAt: now(),
      duration: now() - execution.startedAt,
      result: { totalUsers, changedUsers, inactiveDistribution: distribution }
    }
  catch err:
    update execution set {
      status: 'FAILED',
      completedAt: now(),
      error: `${errorCode}: ${errorMessage}`,  // 拼接结构化前缀
      logs: err.stack
    }
```

**前置条件提示**：M365 Admin Center "Display concealed user/group/site names in all reports" 必须**关闭**，否则第 3 步 CSV 里的 UPN 是 hash，第 4 步 JOIN 全部失败。算法主动检测：首行 UPN 不含 `@` 即抛 `REPORTS_OBFUSCATED`。

#### 实现依赖

- Graph SDK：直接 `fetch` 或复用项目已有的 MsalClient + 手写 `axios` 调用（参考 `entra.service.ts`），不引入 `@microsoft/microsoft-graph-client` 增加体积。
- CSV 解析：`csv-parse`（项目当前未安装，本模块新引入；选 `csv-parse/sync` 子包，TypeScript types 完整、无依赖）。
- License SKU 显示名映射表：`backend/src/modules/ops-center/m365-dormant/sku-display-names.ts`，覆盖常见 SKU（M365 E1/E3/E5、EMS、Defender、PowerBI Pro、Exchange Online 等）。Microsoft 完整对照见 [licensing-service-plan-reference](https://learn.microsoft.com/en-us/azure/active-directory/enterprise-users/licensing-service-plan-reference)，按需补充。

### 中途失败的事务性策略（v1）

同步循环里如果在第 i 个用户出错，前 i-1 个用户的 `M365User` 已经 upsert + `M365UserActivityChange` 已经追加：

- **不回滚**。事件流追加无害；下次同步会覆盖剩余用户的当前态。
- batch 被标 `failed` + `errorCode` + `errorMessage`，前端显示"上次同步失败，部分用户数据可能未刷新"。
- 不在事务里包 Graph 调用（反模式：事务里嵌网络调用会长时间锁表）。
- 直方图聚合发生在最后，failed 批次的 `inactiveDistribution` 为 null（趋势图不画此点）。

### 异常恢复（进程重启 / OOM）

**问题**：后端在 `RUNNING` 中进程崩溃 / 部署，execution 永远卡在 `RUNNING`，service-level 互斥锁拦死下次同步。

**v1 方案**：

**先查现有同步模块（dingtalk / adp / scheduled-sync）是否已有统一的 stuck-execution 清理 hook**——
- 如果有，本模块**直接复用**，不再重复实现。
- 如果没有，本模块在 `OpsCenterM365Module.onModuleInit` 加 startup hook：
  ```ts
  await prisma.automationExecution.updateMany({
    where: {
      task: { type: 'M365_DORMANT_SYNC' },
      status: 'RUNNING',
      startedAt: { lt: subMinutes(new Date(), 30) }
    },
    data: { status: 'TIMEOUT', completedAt: new Date(), error: 'SYNC_INTERRUPTED: Backend restarted while running' }
  })
  ```
- 30 分钟阈值与 PRD §"非功能" 的"单次同步 ≤ 5 分钟"留足缓冲。
- 用 `TIMEOUT` 状态（AutomationExecutionStatus 已有此值），不用自造 `failed`。
- 错误前缀用 `SYNC_INTERRUPTED:`，前端按前缀匹配显示友好文案。
- **防御性**：startup hook 必须包 `try/catch`，失败仅 `logger.warn`，不抛——否则启动期 DB 临时不可用会拖死整个 backend 启动。模块加载顺序也要让 hook 在 PrismaService ready 后才跑。

### Change 事件保留与清理

- v1 **保留全量**，不删除。
- 估算每周新增 100~500 条，1 年 5k~25k 条，5 年 25k~125k 条——单表完全可承受。
- 表上已有 `(userId, createdAt desc)` 复合索引，时间线查询走索引。
- 如未来超过 200k 行：将"超过 N 年的 change 事件归档到冷表 `m365_user_activity_change_archive`"（v2 评估，v1 不做）。

直方图聚合 SQL（PostgreSQL 示例）：

```sql
SELECT
  count(*) FILTER (WHERE last_any_activity IS NULL) AS never,
  count(*) FILTER (WHERE days_inactive >= 0   AND days_inactive < 30)  AS b1,
  count(*) FILTER (WHERE days_inactive >= 30  AND days_inactive < 60)  AS b2,
  count(*) FILTER (WHERE days_inactive >= 60  AND days_inactive < 90)  AS b3,
  count(*) FILTER (WHERE days_inactive >= 90  AND days_inactive < 180) AS b4,
  count(*) FILTER (WHERE days_inactive >= 180 AND days_inactive < 365) AS b5,
  count(*) FILTER (WHERE days_inactive >= 365)                         AS b6
FROM m365_user
WHERE organization_id = $1;
```