diff --git a/make_post_sell/lib/s3_mirror.py b/make_post_sell/lib/s3_mirror.py new file mode 100644 index 0000000..ddd0729 --- /dev/null +++ b/make_post_sell/lib/s3_mirror.py @@ -0,0 +1,306 @@ +"""S3 mirror — fire-and-forget sync of uploaded files to a shop's custom bucket. + +When a shop has mirror_s3_* credentials configured and enabled, every file +written to the MPS main bucket is copied to the shop's bucket in a background +thread. The MPS bucket remains the origin/CDN — the shop bucket is a mirror. +""" + +import fcntl +import logging +import os +import threading + +import boto3 + +log = logging.getLogger(__name__) + + +def _make_mirror_client(endpoint, region, access_key, secret_key): + """Create a boto3 S3 client from mirror credentials.""" + session = boto3.session.Session() + return session.client( + "s3", + region_name=region or "us-east-1", + endpoint_url=endpoint, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + ) + + +def test_mirror_connection(shop): + """Validate shop mirror credentials by listing the bucket. + + Returns (True, "") on success or (False, "error message") on failure. + """ + try: + client = _make_mirror_client( + shop.mirror_s3_endpoint, + shop.mirror_s3_region, + shop.mirror_s3_access_key, + shop.mirror_s3_secret_key, + ) + client.list_objects_v2(Bucket=shop.mirror_s3_bucket, MaxKeys=0) + return True, "" + except Exception as e: + return False, str(e) + + +def mirror_key(src_client, src_bucket, key, dst_client, dst_bucket, + content_type=None, cache_control=None): + """Copy a single key from source bucket to mirror bucket. + + Streams via get_object → put_object. The StreamingBody from get_object + is passed directly as Body to put_object. + """ + try: + resp = src_client.get_object(Bucket=src_bucket, Key=key) + body = resp["Body"] + + put_kwargs = { + "Bucket": dst_bucket, + "Key": key, + "Body": body, + } + if content_type: + put_kwargs["ContentType"] = content_type + if cache_control: + put_kwargs["CacheControl"] = cache_control + + dst_client.put_object(**put_kwargs) + log.info("Mirrored %s to %s/%s", key, dst_bucket, key) + except Exception: + log.exception("Failed to mirror key %s to bucket %s", key, dst_bucket) + + +def _capture_shop_mirror_creds(shop): + """Capture mirror credentials as plain strings for thread safety. + + The shop ORM object must not be accessed from a background thread + after the request's DB session is closed. + """ + return { + "endpoint": shop.mirror_s3_endpoint, + "region": shop.mirror_s3_region, + "bucket": shop.mirror_s3_bucket, + "access_key": shop.mirror_s3_access_key, + "secret_key": shop.mirror_s3_secret_key, + } + + +def _capture_source_creds(src_client): + """Capture source S3 client credentials for thread safety. + + boto3 clients are not thread-safe — the background thread must + create its own client. + """ + return { + "endpoint": src_client._endpoint.host, + "region": src_client.meta.region_name, + "access_key": src_client._request_signer._credentials.access_key, + "secret_key": src_client._request_signer._credentials.secret_key, + } + + +def mirror_key_async(src_client, src_bucket, key, shop, + content_type=None, cache_control=None): + """Fire-and-forget: mirror a key to the shop's bucket in a daemon thread. + + Does nothing if the shop has no mirror configured. + """ + if not shop.has_s3_mirror: + return + + mirror_creds = _capture_shop_mirror_creds(shop) + source_creds = _capture_source_creds(src_client) + + def _sync(): + try: + sess = boto3.session.Session() + src = sess.client( + "s3", + region_name=source_creds["region"], + endpoint_url=source_creds["endpoint"], + aws_access_key_id=source_creds["access_key"], + aws_secret_access_key=source_creds["secret_key"], + ) + dst = _make_mirror_client( + mirror_creds["endpoint"], + mirror_creds["region"], + mirror_creds["access_key"], + mirror_creds["secret_key"], + ) + mirror_key(src, src_bucket, key, dst, mirror_creds["bucket"], + content_type=content_type, cache_control=cache_control) + except Exception: + log.exception("Mirror thread failed for key %s", key) + + t = threading.Thread(target=_sync, daemon=True) + t.start() + + +def mirror_keys_async(src_client, src_bucket, keys, shop, + content_type=None, cache_control=None): + """Fire-and-forget: mirror multiple keys in a single daemon thread.""" + if not shop.has_s3_mirror: + return + + mirror_creds = _capture_shop_mirror_creds(shop) + source_creds = _capture_source_creds(src_client) + + def _sync(): + try: + sess = boto3.session.Session() + src = sess.client( + "s3", + region_name=source_creds["region"], + endpoint_url=source_creds["endpoint"], + aws_access_key_id=source_creds["access_key"], + aws_secret_access_key=source_creds["secret_key"], + ) + dst = _make_mirror_client( + mirror_creds["endpoint"], + mirror_creds["region"], + mirror_creds["access_key"], + mirror_creds["secret_key"], + ) + for key in keys: + mirror_key(src, src_bucket, key, dst, mirror_creds["bucket"], + content_type=content_type, cache_control=cache_control) + except Exception: + log.exception("Mirror thread failed for keys %s", keys) + + t = threading.Thread(target=_sync, daemon=True) + t.start() + + +def backfill_mirror_async(shop_id, session_factory, app_settings): + """Copy all existing S3 files for a shop to the mirror bucket. + + Forks a child process so the backfill survives uWSGI worker recycling. + Uses fcntl.flock on a lockfile for one-per-shop guard. + """ + shop_id_str = str(shop_id) + lockfile = f"/tmp/s3_mirror_backfill_{shop_id_str}.lock" + + db_url = str(session_factory().get_bind().url) + + # Quick check — if lockfile exists and is locked, skip + try: + check_fd = open(lockfile, "w") + fcntl.flock(check_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(check_fd, fcntl.LOCK_UN) + check_fd.close() + except (IOError, OSError): + return # another backfill is running for this shop + + pid = os.fork() + if pid > 0: + os.waitpid(pid, 0) + return + + # Intermediate child: detach fully from uWSGI, then fork again + os.setsid() + pid2 = os.fork() + if pid2 > 0: + os._exit(0) + + # --- Grandchild process: fully detached from uWSGI --- + import resource + maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1] + if maxfd == resource.RLIM_INFINITY: + maxfd = 1024 + for fd in range(3, maxfd): + try: + os.close(fd) + except OSError: + pass + + lock_fd = None + try: + lock_fd = open(lockfile, "w") + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + lock_fd.write(str(os.getpid())) + lock_fd.flush() + + from sqlalchemy import create_engine + from sqlalchemy.orm import Session as SASession + from ..models.shop import Shop + + engine = create_engine(db_url) + session = SASession(bind=engine) + + def _make_src(): + return boto3.session.Session().client( + "s3", + region_name=app_settings["bucket.secure_uploads.region"], + endpoint_url=app_settings["bucket.secure_uploads.post_endpoint"], + aws_access_key_id=app_settings["bucket.secure_uploads.access_key"], + aws_secret_access_key=app_settings["bucket.secure_uploads.secret_key"], + ) + + try: + shop = session.get(Shop, shop_id) + if not shop or not shop.has_s3_mirror: + return + + src = _make_src() + dst = _make_mirror_client( + shop.mirror_s3_endpoint, + shop.mirror_s3_region, + shop.mirror_s3_access_key, + shop.mirror_s3_secret_key, + ) + src_bucket = app_settings["bucket.secure_uploads"] + dst_bucket = shop.mirror_s3_bucket + + # List all objects under the shop's prefix + prefix = f"{shop_id_str}/" + continuation_token = None + total = 0 + mirrored = 0 + + log.info("Backfill mirror for shop %s (pid %d)", shop.name, os.getpid()) + + while True: + list_kwargs = { + "Bucket": src_bucket, + "Prefix": prefix, + "MaxKeys": 1000, + } + if continuation_token: + list_kwargs["ContinuationToken"] = continuation_token + + resp = src.list_objects_v2(**list_kwargs) + contents = resp.get("Contents", []) + total += len(contents) + + for obj in contents: + key = obj["Key"] + mirror_key(src, src_bucket, key, dst, dst_bucket) + mirrored += 1 + + if resp.get("IsTruncated"): + continuation_token = resp["NextContinuationToken"] + else: + break + + log.info("Backfill mirror complete for shop %s: %d/%d keys mirrored", + shop.name, mirrored, total) + + except Exception: + log.exception("Backfill mirror failed for shop %s", shop_id_str) + finally: + session.close() + engine.dispose() + except (IOError, OSError): + pass # couldn't acquire lock + except Exception: + log.exception("Backfill mirror child process failed for shop %s", shop_id_str) + finally: + if lock_fd: + try: + lock_fd.close() + os.unlink(lockfile) + except OSError: + pass + os._exit(0) diff --git a/make_post_sell/models/shop.py b/make_post_sell/models/shop.py index a619fcf..50af0a2 100644 --- a/make_post_sell/models/shop.py +++ b/make_post_sell/models/shop.py @@ -104,6 +104,14 @@ class Shop(RBase, Base): unsandbox_public_key = Column(Unicode(128), nullable=True) unsandbox_secret_key = Column(Unicode(128), nullable=True) + # S3-compatible bucket for mirroring all shop uploads + mirror_s3_endpoint = Column(Unicode(256), nullable=True) + mirror_s3_region = Column(Unicode(64), nullable=True) + mirror_s3_bucket = Column(Unicode(128), nullable=True) + mirror_s3_access_key = Column(Unicode(128), nullable=True) + mirror_s3_secret_key = Column(Unicode(128), nullable=True) + mirror_s3_enabled = Column(Boolean, default=False) + created_timestamp = Column(BigInteger, nullable=False) updated_timestamp = Column(BigInteger, nullable=False) @@ -227,6 +235,17 @@ class Shop(RBase, Base): us.role_id = role_id return us + @property + def has_s3_mirror(self): + """Return True if shop has S3 mirror bucket configured and enabled.""" + return bool( + self.mirror_s3_enabled + and self.mirror_s3_endpoint + and self.mirror_s3_bucket + and self.mirror_s3_access_key + and self.mirror_s3_secret_key + ) + @property def owners(self): """Returns a list of user objects who have the owner role on this shop.""" diff --git a/make_post_sell/scripts/alembic/versions/6b516114c393_add_s3_mirror_bucket_credentials_to_shop.py b/make_post_sell/scripts/alembic/versions/6b516114c393_add_s3_mirror_bucket_credentials_to_shop.py new file mode 100644 index 0000000..87e71e6 --- /dev/null +++ b/make_post_sell/scripts/alembic/versions/6b516114c393_add_s3_mirror_bucket_credentials_to_shop.py @@ -0,0 +1,56 @@ +"""add S3 mirror bucket credentials to shop + +Revision ID: 6b516114c393 +Revises: f898ba460612 +Create Date: 2026-02-27 15:43:58.571158 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '6b516114c393' +down_revision = 'f898ba460612' +branch_labels = None +depends_on = None + +def _column_exists(table, column): + conn = op.get_bind() + result = conn.execute(sa.text(f"PRAGMA table_info({table})")) + return any(row[1] == column for row in result.fetchall()) + + +def upgrade(): + columns = [ + ("mirror_s3_endpoint", sa.Unicode(256), True, None), + ("mirror_s3_region", sa.Unicode(64), True, None), + ("mirror_s3_bucket", sa.Unicode(128), True, None), + ("mirror_s3_access_key", sa.Unicode(128), True, None), + ("mirror_s3_secret_key", sa.Unicode(128), True, None), + ("mirror_s3_enabled", sa.Boolean(), False, "0"), + ] + for col_name, col_type, nullable, server_default in columns: + if not _column_exists("mps_shop", col_name): + op.add_column( + "mps_shop", + sa.Column( + col_name, + col_type, + nullable=nullable, + server_default=server_default, + ), + ) + + +def downgrade(): + for col_name in ( + "mirror_s3_enabled", + "mirror_s3_secret_key", + "mirror_s3_access_key", + "mirror_s3_bucket", + "mirror_s3_region", + "mirror_s3_endpoint", + ): + if _column_exists("mps_shop", col_name): + op.drop_column("mps_shop", col_name) diff --git a/make_post_sell/templates/shop_settings.j2 b/make_post_sell/templates/shop_settings.j2 index 6bd44cd..3ecfb9f 100644 --- a/make_post_sell/templates/shop_settings.j2 +++ b/make_post_sell/templates/shop_settings.j2 @@ -470,6 +470,123 @@

+
+
+ +

S3 Mirror Bucket

+

Connect an S3-compatible bucket to automatically mirror all uploaded files. Works with DigitalOcean Spaces, AWS S3, MinIO, Backblaze B2, and any S3-compatible service. Your MPS bucket remains the origin — this is a backup/mirror.

+ +
+ + +
+ Show Mirror Bucket Settings + +
+ +
+ + + +
+
+ + + + +
+
+ + + + +
+
+ + + + +
+
+ + + + +
+
+ + + + +
+
+ + {% if request.shop.has_s3_mirror %} + Mirror configured and enabled + {% elif request.shop.mirror_s3_bucket %} + Mirror credentials saved but not enabled. Check the box above to activate. + {% else %} + Add S3-compatible bucket credentials to enable file mirroring. + {% endif %} + +
+
+ + + +
+
+
+ + {% if request.shop.has_s3_mirror %} +
+ + +
+ Copy all existing files to your mirror bucket. Runs in background. +
+ {% endif %} + +
+
+ +
+
+ {% if request.monero_enabled %}
diff --git a/make_post_sell/tests/test_functional.py b/make_post_sell/tests/test_functional.py index 50bda3b..de9ca24 100644 --- a/make_post_sell/tests/test_functional.py +++ b/make_post_sell/tests/test_functional.py @@ -4044,3 +4044,113 @@ class TestAnalytics(AuthenticatedFunctionalTests): # Should NOT contain media preview elements self.assertNotIn("edit-media-preview", body) self.assertNotIn("