Merge pull request #735 from JustAJobApp/feat/705-persist-tokens

Feat: Add Premium Tier for Auto Refresh

Author: lnovitz

.github/actions/deploy-backend-to-lightsail/action.yaml

@@ -40,6 +40,9 @@ inputs:
   origin:
     description: Origin
     required: true
+  token-encryption-key:
+    description: Fernet key for encrypting OAuth tokens in DB
+    required: true
 
 runs:
   using: composite
@@ -59,6 +62,7 @@ runs:
       echo "::add-mask::${{ inputs.google-client-redirect-uri }}"
       echo "::add-mask::${{ inputs.ls-database-name }}"
       echo "::add-mask::${{ inputs.origin }}"
+      echo "::add-mask::${{ inputs.token-encryption-key }}"
 
   - name: Set up AWS CLI
     uses: aws-actions/configure-aws-credentials@v1
@@ -188,6 +192,7 @@ runs:
       APP_URL: ${{ inputs.app-url }}
       API_URL: ${{ inputs.api-url }}
       ORIGIN: ${{ inputs.origin }}
+      TOKEN_ENCRYPTION_KEY: ${{ inputs.token-encryption-key }}
     shell: bash
     run: |
       # Mask env vars before use
@@ -201,6 +206,7 @@ runs:
       echo "::add-mask::${DATABASE_URL}"
       echo "::add-mask::${APP_URL}"
       echo "::add-mask::${API_URL}"
+      echo "::add-mask::${TOKEN_ENCRYPTION_KEY}"
 
       # Deploy to Lightsail using inline JSON
       aws lightsail create-container-service-deployment \
@@ -222,7 +228,8 @@ runs:
               \"API_URL\": \"${API_URL}\",
               \"ENV\": \"prod\",
               \"DATABASE_URL\": \"${DATABASE_URL}\",
-              \"ORIGIN\": \"${ORIGIN}\"
+              \"ORIGIN\": \"${ORIGIN}\",
+              \"TOKEN_ENCRYPTION_KEY\": \"${TOKEN_ENCRYPTION_KEY}\"
             },
             \"ports\": {
               \"${CONTAINER_PORT}\": \"HTTP\"

.github/workflows/cd.yml

@@ -34,6 +34,7 @@ jobs:
         google-client-secret: ${{ secrets.GOOGLE_CLIENT_SECRET }}
         google-client-redirect-uri: ${{ secrets.GOOGLE_CLIENT_REDIRECT_URI }}
         origin: ${{ secrets.ORIGIN }}
+        token-encryption-key: ${{ secrets.TOKEN_ENCRYPTION_KEY }}
 
   deploy-frontend:
     name: Deploy Frontend

backend/alembic/versions/add_always_open_fields.py

@@ -0,0 +1,87 @@
+"""add always_open fields for background sync
+
+Revision ID: add_always_open_fields
+Revises: add_applications_found
+Create Date: 2026-02-03
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+# revision identifiers, used by Alembic.
+revision: str = "add_always_open_fields"
+down_revision: Union[str, None] = "add_applications_found"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add always_open fields to users table for background sync feature."""
+
+    # Add sync_tier for tiered scheduling (none/premium)
+    op.add_column(
+        "users",
+        sa.Column(
+            "sync_tier",
+            sa.String(20),
+            nullable=False,
+            server_default="none",
+        ),
+    )
+
+    # Add last_background_sync_at timestamp
+    op.add_column(
+        "users",
+        sa.Column(
+            "last_background_sync_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+    )
+
+    # Create index on sync_tier for efficient batch queries
+    op.create_index(
+        "idx_users_sync_tier",
+        "users",
+        ["sync_tier"],
+        postgresql_where=sa.text("sync_tier = 'premium'"),
+    )
+
+    # Set existing active users to onboarding_completed_at = now
+    # This ensures existing beta users are not forced into the new onboarding flow
+    print("Setting sync_tier = premium for all existing users with active coach link")
+    op.execute(sa.text("UPDATE users SET sync_tier = 'premium' WHERE user_id in (" \
+    "select client_id from coach_client_link where end_date is null)"))
+
+    # Drop payment_asks table - no longer used
+    op.drop_index("idx_payment_asks_user_shown", table_name="payment_asks")
+    op.drop_table("payment_asks")
+
+
+
+def downgrade() -> None:
+    """Remove background sync fields from users table."""
+    # Recreate payment_asks table
+    op.create_table(
+        "payment_asks",
+        sa.Column("id", sa.UUID(), nullable=False, server_default=sa.text("gen_random_uuid()")),
+        sa.Column("user_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("trigger_type", sa.String(100), nullable=False),
+        sa.Column("trigger_value", sa.String(255), nullable=True),
+        sa.Column("shown_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("action", sa.String(50), nullable=True),
+        sa.Column("selected_amount_cents", sa.Integer(), nullable=True),
+        sa.Column("action_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("idx_payment_asks_user_shown", "payment_asks", ["user_id", "shown_at"])
+
+    op.drop_index("idx_users_sync_tier", table_name="users")
+    op.drop_column("users", "last_background_sync_at")
+    op.drop_column("users", "sync_tier")

backend/db/payment_asks.py

@@ -30,6 +30,9 @@ class Users(SQLModel, table=True):
     contribution_started_at: datetime | None = Field(default=None, nullable=True)  # When they first paid
     total_contributed_cents: int = Field(default=0, nullable=False)  # Lifetime contributions
     stripe_subscription_id: str | None = Field(default=None, nullable=True)  # For subscription management
+    # Background sync fields
+    sync_tier: str = Field(default="none", nullable=False)  # 'none' or 'premium'
+    last_background_sync_at: datetime | None = Field(default=None, nullable=True)  # Last background sync timestamp
 
 class CoachClientLink(SQLModel, table=True):
     __tablename__ = "coach_client_link"

backend/db/users.py

@@ -11,14 +11,31 @@
 from utils.config_utils import get_settings
 from contextlib import asynccontextmanager
 import database  # noqa: F401 - used for dependency injection
+from scheduler.background_scheduler import start_scheduler, stop_scheduler
 # Import routes
 from routes import email_routes, auth_routes, file_routes, users_routes, start_date_routes, job_applications_routes, coach_routes, onboarding_routes, stripe_webhook_routes, payment_routes
 
+# Configure logging early so it's available in lifespan
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.DEBUG, format="%(levelname)s - %(message)s")
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # App startup - no dev user seeding (use real OAuth flow for testing)
+    # App startup
+    settings = get_settings()
+
+    # Start background scheduler for Always Open email sync (production only)
+    if settings.is_publicly_deployed:
+        start_scheduler()
+        logger.info("Background scheduler started for Always Open email sync")
+
     yield
 
+    # App shutdown
+    if settings.is_publicly_deployed:
+        stop_scheduler()
+        logger.info("Background scheduler stopped")
+
 app = FastAPI(lifespan=lifespan)
 settings = get_settings()
 APP_URL = settings.APP_URL
@@ -75,10 +92,6 @@ async def lifespan(app: FastAPI):
     )
 
 
-logger = logging.getLogger(__name__)
-logging.basicConfig(level=logging.DEBUG, format="%(levelname)s - %(message)s")
-
-
 # Rate limit exception handler
 @app.exception_handler(RateLimitExceeded)
 async def rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded):

backend/main.py

@@ -1,4 +1,5 @@
 alembic==1.15.1
+APScheduler==3.10.4
 altgraph==0.17.4
 annotated-types==0.7.0
 anyio==4.7.0

backend/requirements.txt

@@ -11,6 +11,7 @@
 from utils.config_utils import get_settings
 from utils.cookie_utils import set_conditional_cookie
 from utils.credential_service import save_credentials
+from utils.billing_utils import is_premium_eligible
 from utils.redirect_utils import Redirects
 from routes.email_routes import fetch_emails_to_db
 import database
@@ -46,8 +47,13 @@ async def login(
     )
     try:
         if not code:
-            # Check if we have a refresh token in session
-            has_refresh_token = get_refresh_token_status(request.session.get("creds"))
+            # Check if we have a refresh token (DB first, then session fallback)
+            session_user_id = request.session.get("user_id")
+            has_refresh_token = get_refresh_token_status(
+                session_creds=request.session.get("creds"),
+                db_session=db_session,
+                user_id=session_user_id,
+            )
             authorization_url, state = get_google_authorization_url(
                 flow, has_refresh_token
             )
@@ -87,11 +93,13 @@ async def login(
         request.session["access_token"] = creds.token
         request.session["creds"] = get_latest_refresh_token(old_creds=request.session.get("creds"), new_creds=creds)
 
-        # Persist encrypted credentials to database for background task support
-        save_credentials(db_session, user.user_id, creds, credential_type="primary")
-
         existing_user, last_fetched_date = user_exists(user, db_session)
-        
+
+        # Only persist credentials to DB for premium users (data minimization)
+        if existing_user and is_premium_eligible(db_session, existing_user):
+            save_credentials(db_session, user.user_id, creds, credential_type="primary")
+            logger.info("Saved credentials for premium user %s", user.user_id)
+
         # Default to False for existing users, will be overwritten if needed
         request.session["is_new_user"] = False 
 
@@ -231,7 +239,13 @@ async def signup(request: Request, db_session: database.DBSession):
     )
     try:
         if not code:
-            has_refresh_token = get_refresh_token_status(request.session.get("creds"))
+            # Check if we have a refresh token (DB first, then session fallback)
+            session_user_id = request.session.get("user_id")
+            has_refresh_token = get_refresh_token_status(
+                session_creds=request.session.get("creds"),
+                db_session=db_session,
+                user_id=session_user_id,
+            )
             authorization_url, state = get_google_authorization_url(
                 flow, has_refresh_token
             )
@@ -337,7 +351,12 @@ async def email_sync_auth(
                 logger.warning("Email sync auth attempted without session. Redirecting to login.")
                 return Redirects.to_error("auth_required")
 
-            has_refresh_token = get_refresh_token_status(request.session.get("email_sync_creds"))
+            # Check for email_sync credentials (DB first, then session fallback)
+            has_refresh_token = get_refresh_token_status(
+                session_creds=request.session.get("email_sync_creds"),
+                db_session=db_session,
+                user_id=user_id,
+            )
             authorization_url, state = get_google_authorization_url(
                 flow, has_refresh_token
             )
@@ -375,12 +394,14 @@ async def email_sync_auth(
         request.session["token_expiry"] = get_token_expiry(creds)
         request.session["access_token"] = creds.token
 
-        # Persist encrypted email_sync credentials to database for background task support
-        save_credentials(db_session, user_id, creds, credential_type="email_sync")
-
         # Update user record with email sync info
         user = db_session.exec(select(Users).where(Users.user_id == user_id)).first()
         if user:
+            # Only persist credentials to DB for premium users (data minimization)
+            if is_premium_eligible(db_session, user):
+                save_credentials(db_session, user_id, creds, credential_type="email_sync")
+                logger.info("Saved email_sync credentials for premium user %s", user_id)
+
             user.has_email_sync_configured = True
             user.sync_email_address = sync_user.user_email
             db_session.add(user)