Feature Flags Explained: Ship Code Safely Without Breaking Production

import json
import os

# ---------------------------------------------------------------------------
# A minimal but realistic feature flag loader.
# In production this would call a service like LaunchDarkly or Unleash,
# but the interface looks identical — that's the whole point.
# ---------------------------------------------------------------------------

class FeatureFlagClient:
    """Reads feature flags from a local JSON config (simulating a flag service)."""

    def __init__(self, config_path: str):
        # Load the flag definitions once at startup.
        # A real SDK would poll or stream updates without a restart.
        with open(config_path, "r") as config_file:
            self._flags = json.load(config_file)

    def is_enabled(self, flag_name: str, user_id: str | None = None) -> bool:
        """Return True if the flag is active for this user."""
        flag = self._flags.get(flag_name)

        if flag is None:
            # Safe default: if the flag doesn't exist, treat it as OFF.
            # Never assume a missing flag means enabled — that's a production incident.
            print(f"[WARN] Flag '{flag_name}' not found. Defaulting to False.")
            return False

        if flag["type"] == "boolean":
            # Simple on/off switch — used for ops flags and full rollouts.
            return flag["enabled"]

        if flag["type"] == "percentage_rollout":
            # Gradually expose a feature to a slice of users.
            # We use a deterministic hash so the same user always gets the same experience.
            if user_id is None:
                return False
            # Stable bucket: hash the user_id into a 0-99 bucket.
            user_bucket = int(user_id.encode().hex(), 16) % 100
            return user_bucket < flag["rollout_percentage"]

        return False  # Unknown flag type — fail safe


# ---------------------------------------------------------------------------
# Simulate a flags.json config that would live in your flag service
# ---------------------------------------------------------------------------
FLAGS_CONFIG = {
    "new_checkout_flow": {
        "type": "percentage_rollout",
        "rollout_percentage": 20   # Only 20% of users see the new flow
    },
    "maintenance_mode": {
        "type": "boolean",
        "enabled": False           # Ops flag — flip this during incidents
    },
    "ai_product_recommendations": {
        "type": "percentage_rollout",
        "rollout_percentage": 5    # Cautious 5% experiment
    }
}

# Write the config to disk so our client can load it
with open("flags.json", "w") as f:
    json.dump(FLAGS_CONFIG, f)

# ---------------------------------------------------------------------------
# Application code — notice how clean this is. No deployment needed to
# change which path a user takes.
# ---------------------------------------------------------------------------
flags = FeatureFlagClient("flags.json")

test_users = ["user_001", "user_042", "user_099", "user_123", "user_777"]

print("=== Checkout Flow Rollout ===")
for user in test_users:
    if flags.is_enabled("new_checkout_flow", user_id=user):
        print(f"  {user} → NEW checkout flow")
    else:
        print(f"  {user} → legacy checkout flow")

print("\n=== Ops Flag Check ===")
if flags.is_enabled("maintenance_mode"):
    print("  Site is in maintenance mode — returning 503")
else:
    print("  Site is operating normally")

print("\n=== Missing Flag (safe default) ===")
result = flags.is_enabled("nonexistent_feature", user_id="user_001")
print(f"  Result: {result}")

# .github/workflows/feature-flag-pipeline.yml
#
# This workflow demonstrates the full lifecycle of a feature flag
# inside a CI/CD pipeline:
#   1. On PR open  → Create the flag in the flag service (defaulting to OFF)
#   2. On push     → Run tests with the flag both ON and OFF
#   3. On PR merge → Schedule the flag for cleanup (prevents flag debt)

name: Feature Flag CI/CD Pipeline

on:
  pull_request:
    types: [opened, synchronize, closed]
    branches:
      - main

env:
  # These would be stored as GitHub Actions secrets in a real repo
  FLAG_SERVICE_API_URL: ${{ secrets.FLAG_SERVICE_API_URL }}
  FLAG_SERVICE_API_KEY: ${{ secrets.FLAG_SERVICE_API_KEY }}

jobs:
  # -----------------------------------------------------------------------
  # JOB 1: When a PR is opened, register the feature flag in your service.
  # This ensures the flag exists before any code runs against it.
  # -----------------------------------------------------------------------
  create-feature-flag:
    name: Register Feature Flag
    runs-on: ubuntu-latest
    if: github.event.action == 'opened'

    steps:
      - name: Derive flag name from branch name
        id: flag_name
        run: |
          # Convert branch name like 'feature/new-checkout-flow'
          # to a safe flag key like 'new_checkout_flow'
          BRANCH_NAME="${{ github.head_ref }}"
          FLAG_KEY=$(echo "$BRANCH_NAME" | sed 's/feature\///' | sed 's/-/_/g')
          echo "flag_key=$FLAG_KEY" >> $GITHUB_OUTPUT
          echo "Derived flag key: $FLAG_KEY"

      - name: Create flag in flag service (defaulting to OFF)
        run: |
          curl -s -X POST "$FLAG_SERVICE_API_URL/flags" \
            -H "Authorization: Bearer $FLAG_SERVICE_API_KEY" \
            -H "Content-Type: application/json" \
            -d '{
              "key": "${{ steps.flag_name.outputs.flag_key }}",
              "name": "Auto-created for PR #${{ github.event.pull_request.number }}",
              "type": "boolean",
              "enabled": false,
              "tags": ["auto-created", "pr-${{ github.event.pull_request.number }}"]
            }'
          echo "Flag '${{ steps.flag_name.outputs.flag_key }}' created and set to OFF"

  # -----------------------------------------------------------------------
  # JOB 2: Run your test suite twice — once with the flag OFF (control)
  # and once with the flag ON (experiment). Both must pass before merge.
  # -----------------------------------------------------------------------
  test-with-flag-variants:
    name: Test Flag ON and OFF Variants
    runs-on: ubuntu-latest
    if: github.event.action == 'synchronize' || github.event.action == 'opened'

    strategy:
      matrix:
        # The matrix runs this job twice in parallel — one per flag state
        flag_state: [enabled, disabled]

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run test suite with flag ${{ matrix.flag_state }}
        env:
          # Inject the flag state as an environment variable.
          # Your app reads this in tests to override the flag service.
          FEATURE_NEW_CHECKOUT_FLOW: ${{ matrix.flag_state == 'enabled' && 'true' || 'false' }}
        run: |
          echo "Running tests with new_checkout_flow = $FEATURE_NEW_CHECKOUT_FLOW"
          python -m pytest tests/ -v --tb=short

  # -----------------------------------------------------------------------
  # JOB 3: When the PR merges, tag the flag for removal.
  # The feature is now fully shipped — the flag is technical debt.
  # -----------------------------------------------------------------------
  schedule-flag-cleanup:
    name: Tag Flag for Removal
    runs-on: ubuntu-latest
    if: github.event.action == 'closed' && github.event.pull_request.merged == true

    steps:
      - name: Tag flag as 'ready-for-removal'
        run: |
          FLAG_KEY=$(echo "${{ github.head_ref }}" | sed 's/feature\///' | sed 's/-/_/g')
          curl -s -X PATCH "$FLAG_SERVICE_API_URL/flags/$FLAG_KEY" \
            -H "Authorization: Bearer $FLAG_SERVICE_API_KEY" \
            -H "Content-Type: application/json" \
            -d '{"tags": ["ready-for-removal", "merged-pr-${{ github.event.pull_request.number }}"] }'
          echo "Flag '$FLAG_KEY' tagged for cleanup. Remove the flag code in the next sprint."

from dataclasses import dataclass
from enum import Enum
import hashlib

# ---------------------------------------------------------------------------
# Modelling the four flag types explicitly.
# Each type has different behaviour, different owners, different lifetimes.
# ---------------------------------------------------------------------------

class FlagType(Enum):
    RELEASE     = "release"      # Engineering owns. Short-lived. Delete after full rollout.
    EXPERIMENT  = "experiment"   # Product owns. Expires when experiment concludes.
    OPS         = "ops"          # SRE owns. Can be permanent. The kill switch.
    PERMISSION  = "permission"   # Platform owns. Permanent. Becomes authorisation logic.


@dataclass
class User:
    user_id: str
    email: str
    subscription_tier: str  # 'free', 'pro', 'enterprise'
    is_internal: bool
    is_beta_tester: bool


class TypingAwareFlagClient:
    """A flag client that evaluates flags differently based on their type."""

    def evaluate_release_flag(self, flag_enabled: bool) -> bool:
        """Release flags are simple booleans. On or off for everyone."""
        return flag_enabled

    def evaluate_experiment_flag(
        self, user: User, experiment_key: str, cohort_percentage: int
    ) -> str:
        """
        Assigns user to a stable experiment cohort.
        Returns 'control' or 'treatment' — never a boolean.
        Experiments have two sides; booleans hide that.
        """
        # Use SHA-256 for a uniform, stable distribution across users.
        # Combining user_id + experiment_key ensures different experiments
        # assign different cohorts to the same user.
        hash_input = f"{user.user_id}:{experiment_key}".encode()
        user_bucket = int(hashlib.sha256(hash_input).hexdigest(), 16) % 100

        return "treatment" if user_bucket < cohort_percentage else "control"

    def evaluate_ops_flag(self, flag_enabled: bool) -> bool:
        """
        Ops flags are deliberate kill switches.
        When enabled=True, the risky behaviour is DISABLED.
        This is intentionally inverted from release flags.
        Enabled means 'the safety net is active'.
        """
        return flag_enabled

    def evaluate_permission_flag(self, user: User, required_tier: str) -> bool:
        """Permission flags check the user's entitlements, not just a boolean."""
        tier_hierarchy = {"free": 0, "pro": 1, "enterprise": 2}
        user_level = tier_hierarchy.get(user.subscription_tier, 0)
        required_level = tier_hierarchy.get(required_tier, 999)
        return user_level >= required_level


# ---------------------------------------------------------------------------
# Putting it all together in a realistic checkout service
# ---------------------------------------------------------------------------
client = TypingAwareFlagClient()

free_user    = User("u_001", "alice@example.com",   "free",       False, False)
pro_user     = User("u_042", "bob@example.com",     "pro",        False, True)
enterprise   = User("u_099", "carol@corp.com",      "enterprise", False, False)
internal_dev = User("u_777", "dave@ourcompany.com", "enterprise", True,  True)

all_users = [free_user, pro_user, enterprise, internal_dev]

# --- 1. RELEASE FLAG: New checkout page is being rolled out ---
new_checkout_enabled = True  # Engineering just flipped this on for 100%
print("=== Release Flag: New Checkout Page ===")
if client.evaluate_release_flag(new_checkout_enabled):
    print("  Serving: new checkout page to all users")
else:
    print("  Serving: legacy checkout page")

# --- 2. EXPERIMENT FLAG: Testing a new 'Free Shipping' banner ---
print("\n=== Experiment Flag: Free Shipping Banner (50% split) ===")
for user in all_users:
    cohort = client.evaluate_experiment_flag(user, "free_shipping_banner_v2", 50)
    print(f"  {user.email:30s} → cohort: {cohort}")

# --- 3. OPS FLAG: Disable AI recommendations under heavy load ---
ai_recommendations_disabled = True  # SRE flipped this during a DB incident
print("\n=== Ops Flag: AI Recommendations Kill Switch ===")
if client.evaluate_ops_flag(ai_recommendations_disabled):
    print("  AI recommendations are DISABLED. Falling back to static popular items.")
else:
    print("  AI recommendations are active.")

# --- 4. PERMISSION FLAG: Advanced analytics dashboard ---
print("\n=== Permission Flag: Advanced Analytics (Pro+ only) ===")
for user in all_users:
    has_access = client.evaluate_permission_flag(user, required_tier="pro")
    status = "GRANTED" if has_access else "denied"
    print(f"  {user.email:30s} ({user.subscription_tier:10s}) → {status}")

from datetime import datetime, timedelta, timezone
from dataclasses import dataclass
from typing import List

# ---------------------------------------------------------------------------
# A flag auditor that runs in CI to catch stale flags before they
# become production liabilities.
#
# In a real system this would pull from your flag service's API.
# Here we simulate the flag registry directly.
# ---------------------------------------------------------------------------

@dataclass
class FlagRecord:
    key: str
    flag_type: str          # 'release', 'experiment', 'ops', 'permission'
    enabled: bool
    rollout_percentage: int  # 0-100
    created_at: datetime
    planned_removal: datetime | None  # None means 'permanent' (ops/permission flags)
    owner_team: str


class FlagDebtAuditor:
    # How long a release flag can be at 100% before it MUST be removed
    RELEASE_FLAG_MAX_LIFETIME_DAYS = 30
    # How long an experiment can run before it's considered abandoned
    EXPERIMENT_MAX_LIFETIME_DAYS = 14

    def __init__(self, flags: List[FlagRecord]):
        self._flags = flags

    def audit(self) -> None:
        now = datetime.now(timezone.utc)
        print("=" * 60)
        print("FEATURE FLAG DEBT AUDIT REPORT")
        print(f"Run at: {now.strftime('%Y-%m-%d %H:%M')} UTC")
        print("=" * 60)

        warnings = []
        errors   = []

        for flag in self._flags:
            age_days = (now - flag.created_at).days

            # --- Check 1: Release flag that's been 100% rolled out too long ---
            if (
                flag.flag_type == "release"
                and flag.rollout_percentage == 100
                and age_days > self.RELEASE_FLAG_MAX_LIFETIME_DAYS
            ):
                errors.append(
                    f"  [ERROR] '{flag.key}' — Release flag at 100% for {age_days} days. "
                    f"Owner: {flag.owner_team}. Remove the flag and its conditional code NOW."
                )

            # --- Check 2: Experiment running too long (probably abandoned) ---
            elif (
                flag.flag_type == "experiment"
                and age_days > self.EXPERIMENT_MAX_LIFETIME_DAYS
            ):
                warnings.append(
                    f"  [WARN]  '{flag.key}' — Experiment flag is {age_days} days old. "
                    f"Owner: {flag.owner_team}. Has the experiment concluded? Clean it up."
                )

            # --- Check 3: Past planned removal date ---
            elif (
                flag.planned_removal is not None
                and now > flag.planned_removal
            ):
                errors.append(
                    f"  [ERROR] '{flag.key}' — Passed planned removal date "
                    f"({flag.planned_removal.strftime('%Y-%m-%d')}). "
                    f"Owner: {flag.owner_team}. This is now flag debt."
                )

            else:
                # Flag looks healthy
                removal_info = (
                    flag.planned_removal.strftime('%Y-%m-%d')
                    if flag.planned_removal else "permanent"
                )
                print(f"  [ OK ]  '{flag.key}' — age: {age_days}d, removal: {removal_info}")

        print()
        for warning in warnings:
            print(warning)
        for error in errors:
            print(error)

        print()
        print(f"Summary: {len(errors)} error(s), {len(warnings)} warning(s)")

        if errors:
            # Fail the CI pipeline if there are flag debt errors
            raise SystemExit(1)


# ---------------------------------------------------------------------------
# Simulated flag registry — this would come from your flag service API
# ---------------------------------------------------------------------------
now = datetime.now(timezone.utc)

flag_registry = [
    FlagRecord(
        key="new_checkout_flow",
        flag_type="release",
        enabled=True,
        rollout_percentage=100,                      # Fully rolled out
        created_at=now - timedelta(days=45),         # But 45 days ago! Too old.
        planned_removal=now - timedelta(days=15),    # Already past due
        owner_team="checkout-team"
    ),
    FlagRecord(
        key="free_shipping_banner_v2",
        flag_type="experiment",
        enabled=True,
        rollout_percentage=50,
        created_at=now - timedelta(days=20),         # Experiment running 20 days
        planned_removal=now + timedelta(days=7),
        owner_team="growth-team"
    ),
    FlagRecord(
        key="ai_recommendations_killswitch",
        flag_type="ops",
        enabled=False,
        rollout_percentage=0,
        created_at=now - timedelta(days=180),        # Old but that's fine for ops flags
        planned_removal=None,                        # Permanent — it's a kill switch
        owner_team="sre-team"
    ),
    FlagRecord(
        key="dark_mode_beta",
        flag_type="release",
        enabled=True,
        rollout_percentage=10,                       # Still in gradual rollout — healthy
        created_at=now - timedelta(days=5),
        planned_removal=now + timedelta(days=25),
        owner_team="ui-team"
    ),
]

auditor = FlagDebtAuditor(flag_registry)
auditor.audit()