feat(loyalty): remind customers before balances expire

Adds what the customer sees of expiry (docs/prd-point-coin.md F6, F12,
PC-504).

GET /customer/wallet/expiring lists everything that will expire, per
currency and day, soonest first. GET /customer/wallet already had the
nearest expiry per currency.

The expiry job now also sends reminders, with the settings of note N4 as
decided: once, reminder_days before (7 by default, 0 for none), per
currency. A customer gets one FCM push per currency and expiry day,
however many lots make it up: "150 EnakPoint akan kedaluwarsa pada 31 Okt
2026. Pakai sebelum hangus.", with type WALLET_EXPIRING, the currency,
amount and expiry_date in its data. Reminders cover whatever falls within
the window, so a run that was missed catches up rather than skipping a day.

Migration 000097 adds wallet_expiry_reminders, one row per customer,
currency and expiry day. The row is written before the push is sent, so
several instances of the job or a restart never remind twice; a push that
then fails is logged and not retried. Lots that expire later on the same
day as an earlier reminder are not reminded of again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/processor/customer_points_processor.go

@@ -224,3 +224,11 @@ func (p *CustomerPointsProcessor) GetFerrisWheelGameAPI(ctx context.Context) (*m
 		},
 	}, nil
 }
+
+func (p *CustomerPointsProcessor) GetCustomerWalletExpiringAPI(ctx context.Context, customerID string) (*models.CustomerWalletExpiringList, error) {
+	id, err := parseWalletCustomerID(customerID)
+	if err != nil {
+		return nil, err
+	}
+	return p.walletQuery.Expiring(ctx, id)
+}

internal/processor/wallet_expiry_processor.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/google/uuid"
 
+	"apskel-pos-be/internal/constants"
 	"apskel-pos-be/internal/logger"
 	"apskel-pos-be/internal/repository"
 )
@@ -23,19 +24,24 @@ const (
 // part of their balance expires.
 const NotificationTypeWalletExpired = "WALLET_EXPIRED"
 
+// NotificationTypeWalletExpiring is the data type of the reminder a customer gets
+// before part of their balance expires.
+const NotificationTypeWalletExpiring = "WALLET_EXPIRING"
+
 // WalletExpiryProcessor takes what is left in lots whose expiry has passed
 // (docs/prd-point-coin.md F12, PC-503). It is safe to run on several instances at
 // once: every lot is expired under its wallet's lock with the key expire:{lot_id}.
 type WalletExpiryProcessor struct {
 	repo     repository.WalletExpiryRepository
+	settings organizationSettingsReader
 	wallet   *WalletProcessor
 	tx       TxRunner
 	notifier customerNotifier
 	now      func() time.Time
 }
 
-func NewWalletExpiryProcessor(repo repository.WalletExpiryRepository, wallet *WalletProcessor, tx TxRunner, notifier customerNotifier) *WalletExpiryProcessor {
-	return &WalletExpiryProcessor{repo: repo, wallet: wallet, tx: tx, notifier: notifier, now: time.Now}
+func NewWalletExpiryProcessor(repo repository.WalletExpiryRepository, settings organizationSettingsReader, wallet *WalletProcessor, tx TxRunner, notifier customerNotifier) *WalletExpiryProcessor {
+	return &WalletExpiryProcessor{repo: repo, settings: settings, wallet: wallet, tx: tx, notifier: notifier, now: time.Now}
 }
 
 type walletExpiredKey struct {
@@ -116,3 +122,75 @@ func expiryDescription(amount int64, currency, sourceDescription string) string
 	}
 	return truncateRunes(description, walletDescriptionLimit)
 }
+
+// SendReminders tells customers, reminder_days before, how much of their balance
+// expires on a day (F12): one push per customer, currency and expiry day, however many
+// lots make it up. A reminder is recorded before it is sent, so another instance or a
+// later run never sends it again; a push that then fails is logged and not retried.
+// It returns how many reminders it sent.
+func (p *WalletExpiryProcessor) SendReminders(ctx context.Context) (int, error) {
+	now := p.now()
+	organizations, err := p.repo.OrganizationsWithUpcomingExpiry(ctx, now)
+	if err != nil {
+		return 0, err
+	}
+	sent := 0
+	for _, organizationID := range organizations {
+		settings, err := p.settings.Organization(ctx, organizationID)
+		if err != nil {
+			logger.NonContext.Error(fmt.Sprintf("Could not read the expiry settings of organization %s; its reminders wait for the next run", organizationID), err)
+			continue
+		}
+		for _, currency := range []string{constants.WalletCurrencyPoint, constants.WalletCurrencyCoin} {
+			days := ExpirySettings(settings, currency).ReminderDays
+			if days <= 0 {
+				continue
+			}
+			until := endOfWalletDay(walletDay(now).AddDate(0, 0, int(days)))
+			upcoming, err := p.repo.UpcomingUnreminded(ctx, organizationID, currency, now, *until)
+			if err != nil {
+				return sent, err
+			}
+			for _, u := range upcoming {
+				first, err := p.repo.MarkReminded(ctx, u, currency)
+				if err != nil {
+					return sent, err
+				}
+				if !first {
+					continue
+				}
+				sent++
+				p.remind(ctx, u, currency)
+			}
+		}
+	}
+	return sent, nil
+}
+
+func (p *WalletExpiryProcessor) remind(ctx context.Context, u repository.UpcomingExpiry, currency string) {
+	if p.notifier == nil {
+		return
+	}
+	name := walletCurrencyName(currency)
+	body := fmt.Sprintf("%d %s akan kedaluwarsa pada %s. Pakai sebelum hangus.", u.Amount, name, formatWalletDate(u.Date))
+	data := map[string]string{
+		"type":        NotificationTypeWalletExpiring,
+		"currency":    currency,
+		"amount":      strconv.FormatInt(u.Amount, 10),
+		"expiry_date": u.Date,
+	}
+	if err := p.notifier.Notify(ctx, u.CustomerID, name+" akan kedaluwarsa", body, data); err != nil {
+		logger.NonContext.Error(fmt.Sprintf("Could not remind customer %s of expiring %s", u.CustomerID, name), err)
+	}
+}
+
+var walletMonthNames = [...]string{"Jan", "Feb", "Mar", "Apr", "Mei", "Jun", "Jul", "Agu", "Sep", "Okt", "Nov", "Des"}
+
+// formatWalletDate writes a YYYY-MM-DD date the way the apps do: "31 Okt 2026".
+func formatWalletDate(date string) string {
+	d, err := time.Parse("2006-01-02", date)
+	if err != nil {
+		return date
+	}
+	return fmt.Sprintf("%d %s %d", d.Day(), walletMonthNames[d.Month()-1], d.Year())
+}

internal/processor/wallet_expiry_processor_test.go

@@ -20,6 +20,8 @@ import (
 type walletExpiryRepoFake struct {
 	wallet *walletRepoFake
 	extra  []repository.DueLot
+	// customer/currency/date of the reminders recorded.
+	reminded map[string]bool
 }
 
 func (f *walletExpiryRepoFake) ListDueLots(_ context.Context, asOf time.Time, limit int) ([]repository.DueLot, error) {
@@ -45,7 +47,7 @@ func (f *walletExpiryRepoFake) ListDueLots(_ context.Context, asOf time.Time, li
 
 func (e *walletMoveEnv) expiry(notifier customerNotifier) (*WalletExpiryProcessor, *walletExpiryRepoFake) {
 	repo := &walletExpiryRepoFake{wallet: e.repo}
-	p := NewWalletExpiryProcessor(repo, e.p, txRunnerFake{}, notifier)
+	p := NewWalletExpiryProcessor(repo, e, e.p, txRunnerFake{}, notifier)
 	p.now = func() time.Time { return e.now }
 	return p, repo
 }
@@ -150,3 +152,93 @@ func TestWalletExpiry_NothingDue(t *testing.T) {
 	assert.Zero(t, count)
 	assert.Empty(t, notifier.pushes)
 }
+
+func (f *walletExpiryRepoFake) OrganizationsWithUpcomingExpiry(_ context.Context, asOf time.Time) ([]uuid.UUID, error) {
+	seen := map[uuid.UUID]bool{}
+	var out []uuid.UUID
+	for _, lot := range f.wallet.lots {
+		if lot.RemainingAmount > 0 && lot.ExpiresAt != nil && lot.ExpiresAt.After(asOf) && !seen[lot.OrganizationID] {
+			seen[lot.OrganizationID] = true
+			out = append(out, lot.OrganizationID)
+		}
+	}
+	return out, nil
+}
+
+func (f *walletExpiryRepoFake) UpcomingUnreminded(_ context.Context, organizationID uuid.UUID, currency string, asOf, until time.Time) ([]repository.UpcomingExpiry, error) {
+	sums := map[[2]string]int64{}
+	var order [][2]string
+	for _, lot := range f.wallet.lots {
+		if lot.OrganizationID != organizationID || lot.Currency != currency || lot.RemainingAmount == 0 ||
+			lot.ExpiresAt == nil || !lot.ExpiresAt.After(asOf) || lot.ExpiresAt.After(until) {
+			continue
+		}
+		key := [2]string{lot.CustomerID.String(), lot.ExpiresAt.In(walletDisplayLocation).Format("2006-01-02")}
+		if f.reminded[key[0]+"/"+currency+"/"+key[1]] {
+			continue
+		}
+		if _, ok := sums[key]; !ok {
+			order = append(order, key)
+		}
+		sums[key] += lot.RemainingAmount
+	}
+	var out []repository.UpcomingExpiry
+	for _, key := range order {
+		out = append(out, repository.UpcomingExpiry{CustomerID: uuid.MustParse(key[0]), Date: key[1], Amount: sums[key]})
+	}
+	return out, nil
+}
+
+func (f *walletExpiryRepoFake) MarkReminded(_ context.Context, u repository.UpcomingExpiry, currency string) (bool, error) {
+	if f.reminded == nil {
+		f.reminded = map[string]bool{}
+	}
+	key := u.CustomerID.String() + "/" + currency + "/" + u.Date
+	if f.reminded[key] {
+		return false, nil
+	}
+	f.reminded[key] = true
+	return true, nil
+}
+
+func TestWalletExpiry_RemindsOncePerDayBeforeExpiry(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	e.now = wib(2026, 10, 25, 9, 0)
+	e.settings.PointExpiry.ReminderDays = 7
+	e.settings.CoinExpiry.ReminderDays = 0 // no reminders for EnakCoin
+	a := e.member("Anita", "081200005678")
+	oct31 := wib(2026, 10, 31, 23, 59)
+	nov30 := wib(2026, 11, 30, 23, 59)
+	e.credit(t, earn(a, 100, &oct31))
+	e.credit(t, earn(a, 50, &oct31))
+	e.credit(t, earn(a, 70, &nov30)) // too far off yet
+	e.earnCoins(t, a, 5, &oct31)
+	notifier := &notifierFake{}
+	p, _ := e.expiry(notifier)
+
+	sent, err := p.SendReminders(e.ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 1, sent)
+	require.Len(t, notifier.pushes[a], 1)
+	push := notifier.pushes[a][0]
+	assert.Equal(t, "EnakPoint akan kedaluwarsa", push.title)
+	assert.Equal(t, "150 EnakPoint akan kedaluwarsa pada 31 Okt 2026. Pakai sebelum hangus.", push.body)
+	assert.Equal(t, map[string]string{"type": NotificationTypeWalletExpiring, "currency": "POINT", "amount": "150", "expiry_date": "2026-10-31"}, push.data)
+
+	// The next run, on this instance or another, sends nothing again.
+	again, err := p.SendReminders(e.ctx)
+	require.NoError(t, err)
+	assert.Zero(t, again)
+
+	// Once 30 Nov comes within seven days, it gets its own reminder.
+	e.now = wib(2026, 11, 23, 9, 0)
+	sent, err = p.SendReminders(e.ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 1, sent)
+	assert.Equal(t, "70", notifier.pushes[a][1].data["amount"])
+}
+
+func TestFormatWalletDate(t *testing.T) {
+	assert.Equal(t, "31 Okt 2026", formatWalletDate("2026-10-31"))
+	assert.Equal(t, "1 Mei 2027", formatWalletDate("2027-05-01"))
+}

internal/processor/wallet_move_db_test.go

@@ -220,7 +220,7 @@ func TestWalletExpiry_TwoInstancesAgainstPostgres(t *testing.T) {
 		wg.Add(1)
 		go func(i int) {
 			defer wg.Done()
-			p := NewWalletExpiryProcessor(repository.NewWalletExpiryRepository(db), wallet, txm, nil)
+			p := NewWalletExpiryProcessor(repository.NewWalletExpiryRepository(db), fixedOrganizationSettings{}, wallet, txm, nil)
 			n, err := p.ExpireDue(context.Background())
 			assert.NoError(t, err)
 			counts[i] = n

internal/processor/wallet_query_processor.go

@@ -315,3 +315,22 @@ func walletTransactionFilter(customerID uuid.UUID, q models.ListCustomerWalletTr
 	}
 	return filter, page, nil
 }
+
+// Expiring is GET /customer/wallet/expiring: what will expire, grouped by day (F6).
+func (p *WalletQueryProcessor) Expiring(ctx context.Context, customerID uuid.UUID) (*models.CustomerWalletExpiringList, error) {
+	rows, err := p.repo.ExpiringByDay(ctx, customerID, p.now())
+	if err != nil {
+		return nil, err
+	}
+	list := &models.CustomerWalletExpiringList{Point: []models.CustomerWalletExpiring{}, Coin: []models.CustomerWalletExpiring{}}
+	for _, row := range rows {
+		item := models.CustomerWalletExpiring{Amount: row.Amount, Date: row.Date}
+		switch row.Currency {
+		case constants.WalletCurrencyPoint:
+			list.Point = append(list.Point, item)
+		case constants.WalletCurrencyCoin:
+			list.Coin = append(list.Coin, item)
+		}
+	}
+	return list, nil
+}

internal/processor/wallet_query_processor_test.go

@@ -243,3 +243,24 @@ func TestWalletQueryProcessor_RejectsBadQueries(t *testing.T) {
 func (f *walletQueryRepoFake) OrganizationOutstanding(context.Context, uuid.UUID) (int64, int64, error) {
 	return 0, 0, nil
 }
+
+func (f *walletQueryRepoFake) ExpiringByDay(context.Context, uuid.UUID, time.Time) ([]repository.WalletExpiringAmount, error) {
+	return f.expiring, nil
+}
+
+func TestWalletQueryProcessor_ExpiringGroupsByCurrencyAndDay(t *testing.T) {
+	repo := &walletQueryRepoFake{org: uuid.New(), expiring: []repository.WalletExpiringAmount{
+		{Currency: "POINT", Date: "2026-10-31", Amount: 150},
+		{Currency: "COIN", Date: "2026-10-31", Amount: 4},
+		{Currency: "POINT", Date: "2026-12-31", Amount: 200},
+	}}
+	got, err := newWalletQueryTest(repo, nil).Expiring(context.Background(), uuid.New())
+	require.NoError(t, err)
+	assert.Equal(t, []models.CustomerWalletExpiring{{Amount: 150, Date: "2026-10-31"}, {Amount: 200, Date: "2026-12-31"}}, got.Point)
+	assert.Equal(t, []models.CustomerWalletExpiring{{Amount: 4, Date: "2026-10-31"}}, got.Coin)
+
+	empty, err := newWalletQueryTest(&walletQueryRepoFake{org: uuid.New()}, nil).Expiring(context.Background(), uuid.New())
+	require.NoError(t, err)
+	assert.NotNil(t, empty.Point, "an empty list, not null")
+	assert.NotNil(t, empty.Coin)
+}