[MM-69215] warn users before their personal access tokens expire (#37272)

* MM-69215: warn users before their personal access tokens expire

Adds a new pat_expiry_notify job that proactively DMs the owner of a
personal access token as it approaches expiry, on a fixed 7/3/1 day
cascade. Previously the only signal was a passive badge in
Profile > Security, so PAT-backed scripts and integrations broke
silently the moment a token lapsed.

Behavior:
- Hourly PeriodicScheduler + SimpleWorker, modeled on expirynotify and
  gated on ServiceSettings.EnableUserAccessTokens.
- Applies to any active token with a non-zero ExpiresAt. Bot-account
  tokens and tokens owned by deactivated users are excluded by the
  store query.
- Delivery is a single system-bot DM (GetSystemBot ->
  GetOrCreateDirectChannel -> CreatePost) localized to the recipient's
  locale, including the token description.
- Only the most urgent applicable bucket is ever sent, so a token that
  first becomes visible already inside the window (short-lived token, or
  a just-reactivated owner) gets a single warning rather than a catch-up
  burst of every threshold it has already passed.

Schema/dedup:
- New nullable UserAccessToken.LastNotifiedThreshold column (migration
  000203). NULL means never notified; it decreases monotonically
  (7 -> 3 -> 1) and prevents the hourly job from re-sending a warning.
- New store methods GetExpiringTokens(now, horizon, limit) and
  UpdateLastNotifiedThreshold, wired through the retry/timer layers and
  mocks.

Covered by unit tests for the bucket math and store query, plus
app-level tests for the dedup, single-warning, and bot-exclusion cases.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: only fetch actionable tokens to avoid notification starvation

Address review feedback on the pat_expiry_notify job.

GetExpiringTokens previously returned every token in the 7-day window
whose marker had not reached the terminal bucket (LastNotifiedThreshold
IS NULL OR > 1), including tokens already warned at their current
bucket. Ordered by ExpiresAt ASC and capped at the batch limit, a
backlog of >1000 already-warned tokens could permanently crowd out a
less-urgent unnotified token past the limit, so it never got warned.

GetExpiringTokens now takes the notification thresholds and returns only
*actionable* rows: for each threshold T, a token qualifies when it has
entered the T-day bucket and has not been notified at T or a more urgent
bucket. Already-warned rows no longer consume a result slot, so a
backlog drains across runs without starving the least-urgent tokens.
The worker keeps recomputing each bucket and re-checking the marker as a
race guard.

Also add a store-test assertion locking down the ExpiresAt-ascending
ordering contract that the worker relies on.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: regenerate store layers and mocks to fix CI drift

The "Check store layers" and "Check mocks" CI jobs failed because the
hand-added GetExpiringTokens and UpdateLastNotifiedThreshold methods were
placed in a different order than the generators emit. Regenerate the
timer/retry layers with `go generate ./channels/store` and reorder the
mock methods to match mockery's alphabetical output. No behavior change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: allow admins to trigger pat_expiry_notify on demand

Add JobTypePatExpiryNotify to the create-job permission allowlist
(PermissionManageJobs), consistent with its siblings expiry_notify and
cleanup_expired_access_tokens. This lets operators run the warning sweep
on demand via POST /api/v4/jobs instead of only on the hourly schedule,
and makes the job creatable from API-level tests.

Add an api4 test covering the new path: a user without manage_jobs is
forbidden, a system admin can create the job. (The notification behavior
itself is covered deterministically by the app-layer tests; the api4
harness does not execute job workers.)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: store last-notified timestamp instead of day bucket

Replace UserAccessToken.LastNotifiedThreshold (the smallest day bucket
already warned) with LastNotifiedAt (the time of the last warning).
Storing the moment rather than a cascade-specific bucket keeps the
column independent of the 7/3/1 day cascade and stays correct if the
buckets are later changed or refined — the bucket already covered is
recovered from (ExpiresAt - LastNotifiedAt).

The actionable-only store query is equally clean: a token still needs a
warning for threshold T when its last warning (if any) was sent while
more than T days remained, i.e. LastNotifiedAt < ExpiresAt - T*day.

Also reword the warning DM to drop the awkward "day(s)" plural: the
7/3-day buckets now render "in N days or less", and the 1-day bucket
uses a dedicated "within 24 hours" message.

Migration 000203 renamed accordingly (column lastnotifiedat bigint);
layers regenerated, mock/tests updated.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: read expiring tokens from master to avoid duplicate warnings

GetExpiringTokens dedups against LastNotifiedAt, which UpdateLastNotifiedAt
writes to master after each warning. Reading the dedup query from a replica
could miss a just-written LastNotifiedAt under replication lag and re-surface
an already-warned token, sending a duplicate DM — a window that on-demand
re-runs of the job (now that the job type is allowlisted) make reachable.
Switch the read to master; it is a once-per-run batch query, so there is no
read-scaling cost.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: shorten GetExpiringTokens master-read comment

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

* MM-69215: point expiry warning at regenerating the token

MM-69561 adds PAT rotation (regenerate a token's secret in place, keeping
its id/description). Reword the pre-expiry warning DMs to tell the owner to
regenerate the token rather than create a new one — regeneration is the
right action for a token that backs an integration, since only the secret
changes. Assert the "regenerate" CTA in the app test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SmTd6i2fqt3YWaaPyorF7

---------

Co-authored-by: Claude <noreply@anthropic.com>

server/channels/api4/job_test.go

@@ -51,6 +51,31 @@ func TestCreateJob(t *testing.T) {
 	})
 }
 
+func TestCreatePatExpiryNotifyJob(t *testing.T) {
+	mainHelper.Parallel(t)
+	th := Setup(t)
+
+	th.LoginSystemManager(t)
+
+	job := &model.Job{Type: model.JobTypePatExpiryNotify}
+
+	t.Run("forbidden without manage_jobs permission", func(t *testing.T) {
+		_, resp, err := th.SystemManagerClient.CreateJob(context.Background(), job)
+		require.Error(t, err)
+		CheckForbiddenStatus(t, resp)
+	})
+
+	t.Run("allowed as system admin", func(t *testing.T) {
+		received, _, err := th.SystemAdminClient.CreateJob(context.Background(), job)
+		require.NoError(t, err)
+		defer func() {
+			result, appErr := th.App.Srv().Store().Job().Delete(received.Id)
+			require.NoErrorf(t, appErr, "Failed to delete job (result: %v): %v", result, appErr)
+		}()
+		require.Equal(t, model.JobTypePatExpiryNotify, received.Type)
+	})
+}
+
 func TestGetJob(t *testing.T) {
 	mainHelper.Parallel(t)
 	th := Setup(t)

server/channels/app/job.go

@@ -224,7 +224,8 @@ func (a *App) SessionHasPermissionToCreateJob(session model.Session, job *model.
 		model.JobTypeExportDelete,
 		model.JobTypeCloud,
 		model.JobTypeExtractContent,
-		model.JobTypeCleanupExpiredAccessTokens:
+		model.JobTypeCleanupExpiredAccessTokens,
+		model.JobTypePatExpiryNotify:
 		return a.SessionHasPermissionTo(session, model.PermissionManageJobs), model.PermissionManageJobs
 	case model.JobTypeAccessControlSync:
 		// Allow system admins to create access control sync jobs

server/channels/app/patexpirynotify.go

@@ -0,0 +1,158 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package app
+
+import (
+	"net/http"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/i18n"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+)
+
+// patExpiryNotifyBatchLimit bounds the number of tokens processed per run.
+// GetExpiringTokens returns only actionable rows (most urgent first), so this
+// caps work per run while a backlog larger than the limit drains across
+// subsequent hourly runs without starving the least-urgent tokens.
+const patExpiryNotifyBatchLimit = 1000
+
+// patExpiryThresholds is the pre-expiry warning cascade, in days, ordered from
+// most to least urgent. A token owner is warned once as the token crosses into
+// each bucket.
+var patExpiryThresholds = []int{1, 3, 7}
+
+// NotifyPersonalAccessTokensExpiring warns the owners of personal access tokens
+// that are approaching expiry, on a fixed 7 / 3 / 1 day cascade. It is invoked
+// hourly by the pat_expiry_notify job.
+//
+// Bot tokens and tokens owned by deactivated users are excluded by the store
+// query. For each remaining token this computes the current warning bucket,
+// sends a single system-bot DM, and records the per-token LastNotifiedAt marker
+// so the same (or a less urgent) warning is never re-sent. Because only
+// the most urgent applicable bucket is ever sent, a token that first becomes
+// visible already inside the window (e.g. created with a short lifetime, or owned
+// by a just-reactivated user) gets a single warning rather than a catch-up burst
+// of every threshold it has already passed.
+func (a *App) NotifyPersonalAccessTokensExpiring() error {
+	if !*a.Config().ServiceSettings.EnableUserAccessTokens {
+		return nil
+	}
+
+	rctx := request.EmptyContext(a.Log().With(mlog.String("component", "pat_expiry_notify")))
+
+	now := model.GetMillis()
+
+	tokens, err := a.Srv().Store().UserAccessToken().GetExpiringTokens(now, patExpiryThresholds, patExpiryNotifyBatchLimit)
+	if err != nil {
+		return model.NewAppError("NotifyPersonalAccessTokensExpiring", "app.user_access_token.get_expiring.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
+	}
+	if len(tokens) == 0 {
+		return nil
+	}
+
+	systemBot, appErr := a.GetSystemBot(rctx)
+	if appErr != nil {
+		return appErr
+	}
+
+	for _, token := range tokens {
+		bucket := patExpiryBucket(token.ExpiresAt, now)
+		if bucket == 0 {
+			// Already expired or outside the cascade; the store window should
+			// prevent this, but guard against clock skew between runs.
+			continue
+		}
+
+		// Skip if the owner has already been warned at this bucket or a more
+		// urgent (smaller) one. A prior warning covers the current bucket when it
+		// was sent while no more time remained than the bucket spans, i.e.
+		// LastNotifiedAt >= ExpiresAt - bucket. nil means no warning sent yet.
+		if token.LastNotifiedAt != nil && *token.LastNotifiedAt >= token.ExpiresAt-int64(bucket)*model.DayInMilliseconds {
+			continue
+		}
+
+		if appErr := a.sendPATExpiryNotification(rctx, systemBot, token, bucket); appErr != nil {
+			rctx.Logger().Error("Failed to send personal access token expiry notification",
+				mlog.String("token_id", token.Id),
+				mlog.String("user_id", token.UserId),
+				mlog.Err(appErr),
+			)
+			continue
+		}
+
+		if err := a.Srv().Store().UserAccessToken().UpdateLastNotifiedAt(token.Id, now); err != nil {
+			rctx.Logger().Error("Failed to update LastNotifiedAt for personal access token",
+				mlog.String("token_id", token.Id),
+				mlog.Err(err),
+			)
+		}
+	}
+
+	return nil
+}
+
+// patExpiryBucket returns the most urgent warning threshold (in days) that the
+// token currently qualifies for, or 0 if it is already expired or further out
+// than the largest threshold. A token with N days remaining maps to the smallest
+// threshold T such that N <= T (e.g. 2 days remaining -> the 3-day bucket).
+func patExpiryBucket(expiresAt int64, now int64) int {
+	remaining := expiresAt - now
+	if remaining <= 0 {
+		return 0
+	}
+	for _, threshold := range patExpiryThresholds {
+		if remaining <= int64(threshold)*model.DayInMilliseconds {
+			return threshold
+		}
+	}
+	return 0
+}
+
+func (a *App) sendPATExpiryNotification(rctx request.CTX, systemBot *model.Bot, token *model.UserAccessToken, bucket int) *model.AppError {
+	channel, appErr := a.GetOrCreateDirectChannel(rctx, token.UserId, systemBot.UserId)
+	if appErr != nil {
+		return appErr
+	}
+
+	user, appErr := a.GetUser(token.UserId)
+	if appErr != nil {
+		return appErr
+	}
+
+	T := i18n.GetUserTranslations(user.Locale)
+
+	description := token.Description
+	if description == "" {
+		description = T("app.pat_expiry_notify.unnamed_token")
+	}
+
+	// The cascade is 7/3/1; the multi-day message is only ever rendered with a
+	// plural day count (7 or 3), and the 1-day bucket uses a dedicated "24 hours"
+	// message, so neither string needs an awkward "day(s)" plural.
+	var message string
+	if bucket <= 1 {
+		message = T("app.pat_expiry_notify.dm_final", map[string]any{
+			"Description": description,
+		})
+	} else {
+		message = T("app.pat_expiry_notify.dm", map[string]any{
+			"Description": description,
+			"Days":        bucket,
+		})
+	}
+
+	post := &model.Post{
+		ChannelId: channel.Id,
+		Message:   message,
+		Type:      model.PostTypeDefault,
+		UserId:    systemBot.UserId,
+	}
+
+	if _, _, err := a.CreatePost(rctx, post, channel, model.CreatePostFlags{SetOnline: false}); err != nil {
+		return err
+	}
+
+	return nil
+}

server/channels/app/patexpirynotify_test.go

@@ -0,0 +1,150 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package app
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost/server/public/model"
+)
+
+func TestPatExpiryBucket(t *testing.T) {
+	now := int64(1_000_000_000_000)
+
+	testCases := []struct {
+		name      string
+		expiresAt int64
+		expected  int
+	}{
+		{"already expired", now - model.DayInMilliseconds, 0},
+		{"expires now", now, 0},
+		{"12 hours left -> 1-day bucket", now + 12*60*60*1000, 1},
+		{"exactly 1 day left -> 1-day bucket", now + 1*model.DayInMilliseconds, 1},
+		{"2 days left -> 3-day bucket", now + 2*model.DayInMilliseconds, 3},
+		{"exactly 3 days left -> 3-day bucket", now + 3*model.DayInMilliseconds, 3},
+		{"5 days left -> 7-day bucket", now + 5*model.DayInMilliseconds, 7},
+		{"exactly 7 days left -> 7-day bucket", now + 7*model.DayInMilliseconds, 7},
+		{"8 days left -> outside cascade", now + 8*model.DayInMilliseconds, 0},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.expected, patExpiryBucket(tc.expiresAt, now))
+		})
+	}
+}
+
+func TestNotifyPersonalAccessTokensExpiring(t *testing.T) {
+	setup := func(t *testing.T) *TestHelper {
+		th := Setup(t).InitBasic(t)
+		th.App.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.EnableUserAccessTokens = true })
+		return th
+	}
+
+	systemBotDMMessages := func(t *testing.T, th *TestHelper, userID string) []string {
+		systemBot, appErr := th.App.GetSystemBot(th.Context)
+		require.Nil(t, appErr)
+		channel, appErr := th.App.GetOrCreateDirectChannel(th.Context, userID, systemBot.UserId)
+		require.Nil(t, appErr)
+		posts, appErr := th.App.GetPosts(th.Context, channel.Id, 0, 50)
+		require.Nil(t, appErr)
+		var messages []string
+		for _, p := range posts.Posts {
+			if p.UserId == systemBot.UserId {
+				messages = append(messages, p.Message)
+			}
+		}
+		return messages
+	}
+
+	t.Run("disabled when EnableUserAccessTokens is off", func(t *testing.T) {
+		th := setup(t)
+		th.App.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.EnableUserAccessTokens = false })
+
+		token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "t", ExpiresAt: model.GetMillis() + 5*model.DayInMilliseconds}
+		_, err := th.App.Srv().Store().UserAccessToken().Save(token)
+		require.NoError(t, err)
+
+		require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring())
+		require.Empty(t, systemBotDMMessages(t, th, th.BasicUser.Id))
+	})
+
+	t.Run("warns once and dedups on re-run", func(t *testing.T) {
+		th := setup(t)
+
+		token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "deploy bot", ExpiresAt: model.GetMillis() + 5*model.DayInMilliseconds}
+		_, err := th.App.Srv().Store().UserAccessToken().Save(token)
+		require.NoError(t, err)
+
+		require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring())
+		messages := systemBotDMMessages(t, th, th.BasicUser.Id)
+		require.Len(t, messages, 1, "owner should get exactly one warning")
+		require.Contains(t, messages[0], "deploy bot", "warning should name the token")
+		require.Contains(t, messages[0], "7 days or less", "5-day token is in the 7-day bucket")
+		require.Contains(t, messages[0], "regenerate", "warning should tell the owner to regenerate the token")
+		require.NotContains(t, messages[0], "day(s)", "message must not use the awkward day(s) plural")
+
+		stored, err := th.App.Srv().Store().UserAccessToken().Get(token.Id)
+		require.NoError(t, err)
+		require.NotNil(t, stored.LastNotifiedAt, "last-notified time should be recorded")
+
+		// A second run in the same bucket must not send another DM.
+		require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring())
+		require.Len(t, systemBotDMMessages(t, th, th.BasicUser.Id), 1, "re-run in the same bucket must not duplicate")
+	})
+
+	t.Run("token already inside window sends only the current bucket, not a catch-up burst", func(t *testing.T) {
+		th := setup(t)
+
+		// Token surfaces with only 2 days left (e.g. created short-lived, or owner
+		// just reactivated). It must get a single warning at the 3-day bucket, never
+		// the already-passed 7-day one.
+		token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "short lived", ExpiresAt: model.GetMillis() + 2*model.DayInMilliseconds}
+		_, err := th.App.Srv().Store().UserAccessToken().Save(token)
+		require.NoError(t, err)
+
+		require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring())
+		messages := systemBotDMMessages(t, th, th.BasicUser.Id)
+		require.Len(t, messages, 1, "exactly one warning, no 7+3 burst")
+		require.Contains(t, messages[0], "3 days or less", "2-day token is in the 3-day bucket")
+
+		stored, err := th.App.Srv().Store().UserAccessToken().Get(token.Id)
+		require.NoError(t, err)
+		require.NotNil(t, stored.LastNotifiedAt, "last-notified time should be recorded")
+	})
+
+	t.Run("sub-day token uses the 24-hour wording", func(t *testing.T) {
+		th := setup(t)
+
+		token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "almost gone", ExpiresAt: model.GetMillis() + 12*60*60*1000}
+		_, err := th.App.Srv().Store().UserAccessToken().Save(token)
+		require.NoError(t, err)
+
+		require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring())
+		messages := systemBotDMMessages(t, th, th.BasicUser.Id)
+		require.Len(t, messages, 1)
+		require.Contains(t, messages[0], "within 24 hours", "sub-day token uses the dedicated 24-hour message")
+		require.NotContains(t, messages[0], "day", "the 24-hour message must not mention days")
+	})
+
+	t.Run("bot-owned token is not notified", func(t *testing.T) {
+		th := setup(t)
+
+		bot, appErr := th.App.CreateBot(th.Context, &model.Bot{Username: "expiring_bot", OwnerId: th.BasicUser.Id})
+		require.Nil(t, appErr)
+
+		token := &model.UserAccessToken{Token: model.NewId(), UserId: bot.UserId, Description: "bot token", ExpiresAt: model.GetMillis() + 5*model.DayInMilliseconds}
+		_, err := th.App.Srv().Store().UserAccessToken().Save(token)
+		require.NoError(t, err)
+
+		require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring())
+		require.Empty(t, systemBotDMMessages(t, th, bot.UserId), "bot accounts must not be warned")
+
+		stored, err := th.App.Srv().Store().UserAccessToken().Get(token.Id)
+		require.NoError(t, err)
+		require.Nil(t, stored.LastNotifiedAt, "bot token marker must stay unset")
+	})
+}

server/channels/app/server.go

@@ -59,6 +59,7 @@ import (
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/migrations"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/mobile_session_metadata"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/notify_admin"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs/pat_expiry_notify"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/plugins"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/post_persistent_notifications"
 	"github.com/mattermost/mattermost/server/v8/channels/jobs/product_notices"
@@ -1747,6 +1748,12 @@ func (s *Server) initJobs() {
 		cleanup_expired_access_tokens.MakeScheduler(s.Jobs),
 	)
 
+	s.Jobs.RegisterJobType(
+		model.JobTypePatExpiryNotify,
+		pat_expiry_notify.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())).NotifyPersonalAccessTokensExpiring),
+		pat_expiry_notify.MakeScheduler(s.Jobs),
+	)
+
 	s.Jobs.RegisterJobType(
 		model.JobTypeRefreshMaterializedViews,
 		refresh_materialized_views.MakeWorker(s.Jobs, *s.platform.Config().SqlSettings.DriverName),

server/channels/db/migrations/migrations.list

@@ -399,3 +399,5 @@ channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_i
 channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_id_index.up.sql
 channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.down.sql
 channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.up.sql
+channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.down.sql
+channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.up.sql

server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.down.sql

@@ -0,0 +1 @@
+ALTER TABLE useraccesstokens DROP COLUMN IF EXISTS lastnotifiedat;

server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.up.sql

@@ -0,0 +1 @@
+ALTER TABLE useraccesstokens ADD COLUMN IF NOT EXISTS lastnotifiedat bigint;

server/channels/jobs/pat_expiry_notify/scheduler.go

@@ -0,0 +1,20 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package pat_expiry_notify
+
+import (
+	"time"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+)
+
+const schedFreq = 1 * time.Hour
+
+func MakeScheduler(jobServer *jobs.JobServer) *jobs.PeriodicScheduler {
+	isEnabled := func(cfg *model.Config) bool {
+		return *cfg.ServiceSettings.EnableUserAccessTokens
+	}
+	return jobs.NewPeriodicScheduler(jobServer, model.JobTypePatExpiryNotify, schedFreq, isEnabled)
+}

server/channels/jobs/pat_expiry_notify/worker.go

@@ -0,0 +1,27 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package pat_expiry_notify
+
+import (
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
+)
+
+// MakeWorker creates a worker that warns the owners of personal access tokens
+// that are approaching expiry. The actual notification logic lives in the app
+// layer and is passed in as notifyExpiring, mirroring the expirynotify job.
+func MakeWorker(jobServer *jobs.JobServer, notifyExpiring func() error) *jobs.SimpleWorker {
+	const workerName = "PatExpiryNotify"
+
+	isEnabled := func(cfg *model.Config) bool {
+		return *cfg.ServiceSettings.EnableUserAccessTokens
+	}
+	execute := func(logger mlog.LoggerIFace, job *model.Job) error {
+		defer jobServer.HandleJobPanic(logger, job)
+
+		return notifyExpiring()
+	}
+	return jobs.NewSimpleWorker(workerName, jobServer, execute, isEnabled)
+}

server/channels/store/retrylayer/retrylayer.go

@@ -18252,6 +18252,27 @@ func (s *RetryLayerUserAccessTokenStore) GetExpiredBefore(cutoff int64, limit in
 
 }
 
+func (s *RetryLayerUserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) {
+
+	tries := 0
+	for {
+		result, err := s.UserAccessTokenStore.GetExpiringTokens(now, thresholds, limit)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerUserAccessTokenStore) Save(token *model.UserAccessToken) (*model.UserAccessToken, error) {
 
 	tries := 0
@@ -18294,6 +18315,27 @@ func (s *RetryLayerUserAccessTokenStore) Search(term string) ([]*model.UserAcces
 
 }
 
+func (s *RetryLayerUserAccessTokenStore) UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error {
+
+	tries := 0
+	for {
+		err := s.UserAccessTokenStore.UpdateLastNotifiedAt(tokenID, notifiedAt)
+		if err == nil {
+			return nil
+		}
+		if !isRepeatableError(err) {
+			return err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerUserAccessTokenStore) UpdateTokenDisable(tokenID string) error {
 
 	tries := 0

server/channels/store/sqlstore/user_access_token_store.go

@@ -33,6 +33,7 @@ func newSqlUserAccessTokenStore(sqlStore *SqlStore) store.UserAccessTokenStore {
 			"UserAccessTokens.Description",
 			"UserAccessTokens.IsActive",
 			"UserAccessTokens.ExpiresAt",
+			"UserAccessTokens.LastNotifiedAt",
 		).
 		From("UserAccessTokens")
 
@@ -270,6 +271,89 @@ func (s SqlUserAccessTokenStore) GetExpiredBefore(cutoff int64, limit int) ([]*m
 	return tokens, nil
 }
 
+// GetExpiringTokens returns active, non-bot tokens belonging to non-deactivated
+// users that need a pre-expiry warning for one of the given day thresholds
+// (e.g. 7/3/1), ordered most-urgent first, up to the given limit. Bot tokens are
+// excluded because bot accounts are exempt from the expiry policy and have no
+// human inbox to notify.
+//
+// Only *actionable* rows are returned: for each threshold T a token qualifies
+// when it has entered the T-day bucket (ExpiresAt <= now + T days) and has not
+// yet been notified at T or a more urgent bucket. "Not yet notified at T" means
+// the last warning (if any) was sent while more than T days remained, i.e.
+// LastNotifiedAt < ExpiresAt - T days (or LastNotifiedAt IS NULL). OR-ing this
+// across every threshold yields exactly the tokens whose current (most urgent)
+// bucket is still un-notified, so tokens already warned at their current bucket
+// never consume a result slot — without this, a backlog of already-warned tokens
+// ordered ahead of a less-urgent unnotified token could starve it past the limit
+// on every run. The worker still recomputes each token's bucket and re-checks the
+// last-notified time as a guard against races. The secret Token column is
+// intentionally NOT selected — callers use the returned rows for metadata
+// (notification, marker update) only.
+//
+// A non-positive limit or empty thresholds returns an empty slice without
+// hitting the DB rather than relying on the int -> uint64 cast (which would
+// otherwise wrap a negative value into an enormous unsigned limit and
+// effectively disable the bound).
+func (s SqlUserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) {
+	tokens := []*model.UserAccessToken{}
+
+	if limit <= 0 || len(thresholds) == 0 {
+		return tokens, nil
+	}
+
+	actionable := sq.Or{}
+	for _, t := range thresholds {
+		bucketMillis := int64(t) * model.DayInMilliseconds
+		actionable = append(actionable, sq.And{
+			sq.LtOrEq{"UserAccessTokens.ExpiresAt": now + bucketMillis},
+			sq.Or{
+				sq.Eq{"UserAccessTokens.LastNotifiedAt": nil},
+				sq.Expr("UserAccessTokens.LastNotifiedAt < UserAccessTokens.ExpiresAt - ?", bucketMillis),
+			},
+		})
+	}
+
+	query := s.getQueryBuilder().
+		Select(
+			"UserAccessTokens.Id",
+			"UserAccessTokens.UserId",
+			"UserAccessTokens.Description",
+			"UserAccessTokens.IsActive",
+			"UserAccessTokens.ExpiresAt",
+			"UserAccessTokens.LastNotifiedAt",
+		).
+		From("UserAccessTokens").
+		InnerJoin("Users ON Users.Id = UserAccessTokens.UserId").
+		LeftJoin("Bots ON Bots.UserId = UserAccessTokens.UserId").
+		Where(sq.Eq{"UserAccessTokens.IsActive": true}).
+		Where(sq.Gt{"UserAccessTokens.ExpiresAt": now}).
+		Where(sq.Eq{"Users.DeleteAt": 0}).
+		Where(sq.Eq{"Bots.UserId": nil}).
+		Where(actionable).
+		OrderBy("UserAccessTokens.ExpiresAt ASC").
+		Limit(uint64(limit))
+
+	// Read from master: this dedups against LastNotifiedAt (written to master),
+	// so a lagging replica could re-surface an already-warned token and send a
+	// duplicate. It runs at most once per job run, so master costs nothing here.
+	if err := s.GetMaster().SelectBuilder(&tokens, query); err != nil {
+		return nil, errors.Wrap(err, "failed to find expiring UserAccessTokens")
+	}
+
+	return tokens, nil
+}
+
+// UpdateLastNotifiedAt records the time (Unix milliseconds) at which the token
+// owner was most recently warned about the token's upcoming expiry, so the hourly
+// pat_expiry_notify job does not re-send the same warning on subsequent runs.
+func (s SqlUserAccessTokenStore) UpdateLastNotifiedAt(tokenId string, notifiedAt int64) error {
+	if _, err := s.GetMaster().Exec("UPDATE UserAccessTokens SET LastNotifiedAt = ? WHERE Id = ?", notifiedAt, tokenId); err != nil {
+		return errors.Wrapf(err, "failed to update LastNotifiedAt for UserAccessToken with id=%s", tokenId)
+	}
+	return nil
+}
+
 // CountNonCompliantExpiry returns the number of active, non-bot tokens that
 // violate the maximum lifetime policy implied by maxExpiresAt. It is used to
 // preview the blast radius before revoking.

server/channels/store/store.go

@@ -853,11 +853,13 @@ type UserAccessTokenStore interface {
 	GetByToken(tokenString string) (*model.UserAccessToken, error)
 	GetByUser(userID string, page, perPage int) ([]*model.UserAccessToken, error)
 	GetExpiredBefore(cutoff int64, limit int) ([]*model.UserAccessToken, error)
+	GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error)
 	CountNonCompliantExpiry(maxExpiresAt int64) (int64, error)
 	DeleteNonCompliantExpiry(maxExpiresAt int64, limit int) ([]string, error)
 	Search(term string) ([]*model.UserAccessToken, error)
 	UpdateTokenEnable(tokenID string) error
 	UpdateTokenDisable(tokenID string) error
+	UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error
 }
 
 type PluginStore interface {

server/channels/store/storetest/mocks/UserAccessTokenStore.go

@@ -286,6 +286,36 @@ func (_m *UserAccessTokenStore) GetExpiredBefore(cutoff int64, limit int) ([]*mo
 	return r0, r1
 }
 
+// GetExpiringTokens provides a mock function with given fields: now, thresholds, limit
+func (_m *UserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) {
+	ret := _m.Called(now, thresholds, limit)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetExpiringTokens")
+	}
+
+	var r0 []*model.UserAccessToken
+	var r1 error
+	if rf, ok := ret.Get(0).(func(int64, []int, int) ([]*model.UserAccessToken, error)); ok {
+		return rf(now, thresholds, limit)
+	}
+	if rf, ok := ret.Get(0).(func(int64, []int, int) []*model.UserAccessToken); ok {
+		r0 = rf(now, thresholds, limit)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.UserAccessToken)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(int64, []int, int) error); ok {
+		r1 = rf(now, thresholds, limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // Save provides a mock function with given fields: token
 func (_m *UserAccessTokenStore) Save(token *model.UserAccessToken) (*model.UserAccessToken, error) {
 	ret := _m.Called(token)
@@ -346,6 +376,24 @@ func (_m *UserAccessTokenStore) Search(term string) ([]*model.UserAccessToken, e
 	return r0, r1
 }
 
+// UpdateLastNotifiedAt provides a mock function with given fields: tokenID, notifiedAt
+func (_m *UserAccessTokenStore) UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error {
+	ret := _m.Called(tokenID, notifiedAt)
+
+	if len(ret) == 0 {
+		panic("no return value specified for UpdateLastNotifiedAt")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, int64) error); ok {
+		r0 = rf(tokenID, notifiedAt)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // UpdateTokenDisable provides a mock function with given fields: tokenID
 func (_m *UserAccessTokenStore) UpdateTokenDisable(tokenID string) error {
 	ret := _m.Called(tokenID)

server/channels/store/storetest/user_access_token_store.go

@@ -19,6 +19,8 @@ func TestUserAccessTokenStore(t *testing.T, rctx request.CTX, ss store.Store) {
 	t.Run("UserAccessTokenSearch", func(t *testing.T) { testUserAccessTokenSearch(t, rctx, ss) })
 	t.Run("UserAccessTokenPagination", func(t *testing.T) { testUserAccessTokenPagination(t, rctx, ss) })
 	t.Run("UserAccessTokenExpiry", func(t *testing.T) { testUserAccessTokenExpiry(t, rctx, ss) })
+	t.Run("UserAccessTokenGetExpiring", func(t *testing.T) { testUserAccessTokenGetExpiring(t, rctx, ss) })
+	t.Run("UserAccessTokenUpdateLastNotifiedAt", func(t *testing.T) { testUserAccessTokenUpdateLastNotifiedAt(t, rctx, ss) })
 	t.Run("UserAccessTokenNonCompliant", func(t *testing.T) { testUserAccessTokenNonCompliant(t, rctx, ss) })
 }
 
@@ -358,6 +360,167 @@ func testUserAccessTokenExpiry(t *testing.T, rctx request.CTX, ss store.Store) {
 	require.Equal(t, int64(0), deleted)
 }
 
+func testUserAccessTokenGetExpiring(t *testing.T, rctx request.CTX, ss store.Store) {
+	now := model.GetMillis()
+	thresholds := []int{1, 3, 7}
+
+	activeUser, err := ss.User().Save(rctx, &model.User{Email: MakeEmail(), Username: model.NewUsername()})
+	require.NoError(t, err)
+
+	// Token expiring inside the 7-day window, never notified.
+	inWindow := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "in-window", ExpiresAt: now + 5*dayMillisTest}
+	_, err = ss.UserAccessToken().Save(inWindow)
+	require.NoError(t, err)
+
+	// Token expiring beyond the horizon.
+	beyond := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "beyond", ExpiresAt: now + 30*dayMillisTest}
+	_, err = ss.UserAccessToken().Save(beyond)
+	require.NoError(t, err)
+
+	// Already-expired token.
+	past := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "past", ExpiresAt: now - 60*1000}
+	_, err = ss.UserAccessToken().Save(past)
+	require.NoError(t, err)
+
+	// Non-expiring token.
+	nonExpiring := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "non-expiring"}
+	_, err = ss.UserAccessToken().Save(nonExpiring)
+	require.NoError(t, err)
+
+	// Token with 12h left that was already warned within the last day (i.e. at the
+	// terminal 1-day bucket): excluded.
+	terminal := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "terminal", ExpiresAt: now + 12*60*60*1000}
+	_, err = ss.UserAccessToken().Save(terminal)
+	require.NoError(t, err)
+	require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(terminal.Id, now))
+
+	// Token warned at the 7-day mark (when 7 days remained) but now inside the
+	// 3-day window: still returned, because the 3-day bucket hasn't been covered.
+	stillNotify := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "still-notify", ExpiresAt: now + 2*dayMillisTest}
+	_, err = ss.UserAccessToken().Save(stillNotify)
+	require.NoError(t, err)
+	stillNotifyAt := stillNotify.ExpiresAt - 7*dayMillisTest
+	require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(stillNotify.Id, stillNotifyAt))
+
+	// Deactivated user with an in-window token: excluded.
+	deletedUser := &model.User{Email: MakeEmail(), Username: model.NewUsername()}
+	_, err = ss.User().Save(rctx, deletedUser)
+	require.NoError(t, err)
+	deletedUser.DeleteAt = model.GetMillis()
+	_, err = ss.User().Update(rctx, deletedUser, true)
+	require.NoError(t, err)
+	deactivatedToken := &model.UserAccessToken{Token: model.NewId(), UserId: deletedUser.Id, Description: "deactivated", ExpiresAt: now + 5*dayMillisTest}
+	_, err = ss.UserAccessToken().Save(deactivatedToken)
+	require.NoError(t, err)
+
+	// Bot with an in-window token: excluded.
+	botUser, err := ss.User().Save(rctx, model.UserFromBot(&model.Bot{Username: model.NewUsername(), OwnerId: activeUser.Id}))
+	require.NoError(t, err)
+	_, nErr := ss.Bot().Save(&model.Bot{UserId: botUser.Id, Username: botUser.Username, OwnerId: activeUser.Id})
+	require.NoError(t, nErr)
+	botToken := &model.UserAccessToken{Token: model.NewId(), UserId: botUser.Id, Description: "bot", ExpiresAt: now + 5*dayMillisTest}
+	_, err = ss.UserAccessToken().Save(botToken)
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		for _, id := range []string{inWindow.Id, beyond.Id, past.Id, nonExpiring.Id, terminal.Id, stillNotify.Id, deactivatedToken.Id, botToken.Id} {
+			_ = ss.UserAccessToken().Delete(id)
+		}
+		_ = ss.Bot().PermanentDelete(botUser.Id)
+		_ = ss.User().PermanentDelete(rctx, botUser.Id)
+		_ = ss.User().PermanentDelete(rctx, activeUser.Id)
+		_ = ss.User().PermanentDelete(rctx, deletedUser.Id)
+	})
+
+	rows, err := ss.UserAccessToken().GetExpiringTokens(now, thresholds, 100)
+	require.NoError(t, err)
+
+	got := make(map[string]*model.UserAccessToken)
+	for _, row := range rows {
+		got[row.Id] = row
+		require.Empty(t, row.Token, "GetExpiringTokens must never return the secret Token value")
+	}
+
+	require.Contains(t, got, inWindow.Id, "in-window token should be returned")
+	require.Contains(t, got, stillNotify.Id, "token warned at 7 days but now inside the 3-day window should still be returned")
+	require.NotContains(t, got, beyond.Id, "token beyond the horizon must not be returned")
+	require.NotContains(t, got, past.Id, "already-expired token must not be returned")
+	require.NotContains(t, got, nonExpiring.Id, "non-expiring token must not be returned")
+	require.NotContains(t, got, terminal.Id, "token already warned within the terminal bucket must not be returned")
+	require.NotContains(t, got, deactivatedToken.Id, "token owned by a deactivated user must not be returned")
+	require.NotContains(t, got, botToken.Id, "bot token must not be returned")
+
+	require.Nil(t, got[inWindow.Id].LastNotifiedAt)
+	require.NotNil(t, got[stillNotify.Id].LastNotifiedAt)
+	require.Equal(t, stillNotifyAt, *got[stillNotify.Id].LastNotifiedAt)
+
+	// Results must be ordered by ExpiresAt ascending (most urgent first); the
+	// worker relies on this to drain a batch in expiry order. stillNotify (2d)
+	// must come before inWindow (5d), and the slice must be non-decreasing.
+	var prev int64
+	for i, row := range rows {
+		if i > 0 {
+			require.GreaterOrEqual(t, row.ExpiresAt, prev, "GetExpiringTokens must return rows in ExpiresAt ascending order")
+		}
+		prev = row.ExpiresAt
+	}
+	require.Less(t, indexOfToken(rows, stillNotify.Id), indexOfToken(rows, inWindow.Id), "sooner-expiring token must be returned first")
+
+	// Non-positive limit and empty thresholds short-circuit.
+	zero, err := ss.UserAccessToken().GetExpiringTokens(now, thresholds, 0)
+	require.NoError(t, err)
+	require.Empty(t, zero)
+	neg, err := ss.UserAccessToken().GetExpiringTokens(now, thresholds, -3)
+	require.NoError(t, err)
+	require.Empty(t, neg)
+	none, err := ss.UserAccessToken().GetExpiringTokens(now, nil, 100)
+	require.NoError(t, err)
+	require.Empty(t, none)
+}
+
+func indexOfToken(tokens []*model.UserAccessToken, id string) int {
+	for i, token := range tokens {
+		if token.Id == id {
+			return i
+		}
+	}
+	return -1
+}
+
+func testUserAccessTokenUpdateLastNotifiedAt(t *testing.T, rctx request.CTX, ss store.Store) {
+	token := &model.UserAccessToken{
+		Token:       model.NewId(),
+		UserId:      model.NewId(),
+		Description: "notified-at",
+		ExpiresAt:   model.GetMillis() + dayMillisTest,
+	}
+	_, err := ss.UserAccessToken().Save(token)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = ss.UserAccessToken().Delete(token.Id) })
+
+	// A freshly saved token has no last-notified time set (NULL -> nil).
+	stored, err := ss.UserAccessToken().Get(token.Id)
+	require.NoError(t, err)
+	require.Nil(t, stored.LastNotifiedAt)
+
+	first := model.GetMillis()
+	require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(token.Id, first))
+	stored, err = ss.UserAccessToken().Get(token.Id)
+	require.NoError(t, err)
+	require.NotNil(t, stored.LastNotifiedAt)
+	require.Equal(t, first, *stored.LastNotifiedAt)
+
+	// A later warning advances the timestamp.
+	second := first + dayMillisTest
+	require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(token.Id, second))
+	stored, err = ss.UserAccessToken().Get(token.Id)
+	require.NoError(t, err)
+	require.NotNil(t, stored.LastNotifiedAt)
+	require.Equal(t, second, *stored.LastNotifiedAt)
+}
+
+const dayMillisTest = int64(24 * 60 * 60 * 1000)
+
 func testUserAccessTokenNonCompliant(t *testing.T, rctx request.CTX, ss store.Store) {
 	now := model.GetMillis()
 	day := int64(24 * 60 * 60 * 1000)

server/channels/store/timerlayer/timerlayer.go

@@ -14417,6 +14417,22 @@ func (s *TimerLayerUserAccessTokenStore) GetExpiredBefore(cutoff int64, limit in
 	return result, err
 }
 
+func (s *TimerLayerUserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) {
+	start := time.Now()
+
+	result, err := s.UserAccessTokenStore.GetExpiringTokens(now, thresholds, limit)
+
+	elapsed := float64(time.Since(start)) / float64(time.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("UserAccessTokenStore.GetExpiringTokens", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerUserAccessTokenStore) Save(token *model.UserAccessToken) (*model.UserAccessToken, error) {
 	start := time.Now()
 
@@ -14449,6 +14465,22 @@ func (s *TimerLayerUserAccessTokenStore) Search(term string) ([]*model.UserAcces
 	return result, err
 }
 
+func (s *TimerLayerUserAccessTokenStore) UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error {
+	start := time.Now()
+
+	err := s.UserAccessTokenStore.UpdateLastNotifiedAt(tokenID, notifiedAt)
+
+	elapsed := float64(time.Since(start)) / float64(time.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("UserAccessTokenStore.UpdateLastNotifiedAt", success, elapsed)
+	}
+	return err
+}
+
 func (s *TimerLayerUserAccessTokenStore) UpdateTokenDisable(tokenID string) error {
 	start := time.Now()
 

server/i18n/en.json

@@ -7912,6 +7912,18 @@
     "id": "app.pap.update_access_control_policies_active.app_error",
     "translation": "Could not update active status of access control policies."
   },
+  {
+    "id": "app.pat_expiry_notify.dm",
+    "translation": "Your personal access token \"{{.Description}}\" will expire in {{.Days}} days or less. Open **Profile > Security** to regenerate it before it stops working."
+  },
+  {
+    "id": "app.pat_expiry_notify.dm_final",
+    "translation": "Your personal access token \"{{.Description}}\" will expire within 24 hours. Open **Profile > Security** to regenerate it before it stops working."
+  },
+  {
+    "id": "app.pat_expiry_notify.unnamed_token",
+    "translation": "(no description)"
+  },
   {
     "id": "app.pdp.access_evaluation.app_error",
     "translation": "Failed evaluate access control policy."
@@ -9698,6 +9710,10 @@
     "id": "app.user_access_token.get_by_user.app_error",
     "translation": "Unable to get the personal access tokens by user."
   },
+  {
+    "id": "app.user_access_token.get_expiring.app_error",
+    "translation": "Unable to retrieve expiring personal access tokens."
+  },
   {
     "id": "app.user_access_token.invalid_or_missing",
     "translation": "Invalid or missing token."

server/public/model/job.go

@@ -49,6 +49,7 @@ const (
 	JobTypeDeleteExpiredPosts            = "delete_expired_posts"
 	JobTypeAutoTranslationRecovery       = "autotranslation_recovery"
 	JobTypeCleanupExpiredAccessTokens    = "cleanup_expired_access_tokens"
+	JobTypePatExpiryNotify               = "pat_expiry_notify"
 
 	JobStatusPending         = "pending"
 	JobStatusInProgress      = "in_progress"
@@ -80,6 +81,7 @@ var AllJobTypes = [...]string{
 	JobTypeLastAccessibleFile,
 	JobTypeCleanupDesktopTokens,
 	JobTypeCleanupExpiredAccessTokens,
+	JobTypePatExpiryNotify,
 	JobTypeRefreshMaterializedViews,
 	JobTypeMobileSessionMetadata,
 }

server/public/model/user_access_token.go

@@ -25,6 +25,15 @@ type UserAccessToken struct {
 	// ExpiresAt is non-zero and in the past are considered expired and
 	// MUST be rejected at validation time.
 	ExpiresAt int64 `json:"expires_at"`
+	// LastNotifiedAt is the Unix timestamp in milliseconds at which the token
+	// owner was last warned that the token is approaching expiry. It is nil until
+	// the first warning is sent. The pat_expiry_notify job uses it to dedup: the
+	// warning bucket already covered is recovered from (ExpiresAt - LastNotifiedAt),
+	// so the same (or a less urgent) warning is not re-sent every run. Storing the
+	// moment rather than the bucket keeps the column independent of the day cascade
+	// and correct if the buckets are later changed. It is internal server
+	// bookkeeping that is never exposed over the API.
+	LastNotifiedAt *int64 `json:"-"`
 }
 
 func (t *UserAccessToken) IsValid() *AppError {