Connect MySQL to PostPenguin

Store PostPenguin blog posts directly in MySQL or MariaDB. Works with any backend that can connect to MySQL.

🗄️ Database Schema

CREATE TABLE posts (
    id INT AUTO_INCREMENT PRIMARY KEY,
    postpenguin_id VARCHAR(255) UNIQUE NOT NULL,
    title VARCHAR(500) NOT NULL,
    slug VARCHAR(255) UNIQUE NOT NULL,
    html LONGTEXT NOT NULL,
    meta_title VARCHAR(500),
    meta_description TEXT,
    featured_image VARCHAR(1000),
    tags JSON,
    status VARCHAR(50) DEFAULT 'publish',
    published_at DATETIME,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    
    INDEX idx_status (status),
    INDEX idx_published_at (published_at),
    INDEX idx_slug (slug)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

🚀 Node.js Webhook Handler

// webhook-handler.js
const express = require('express')
const mysql = require('mysql2/promise')
const crypto = require('crypto')

const app = express()
app.use(express.json())

// MySQL connection pool
const pool = mysql.createPool({
    host: process.env.MYSQL_HOST || 'localhost',
    user: process.env.MYSQL_USER,
    password: process.env.MYSQL_PASSWORD,
    database: process.env.MYSQL_DATABASE,
    waitForConnections: true,
    connectionLimit: 10,
})

// Verify webhook signature
function verifySignature(payload, signature, secret) {
    const expected = crypto
        .createHmac('sha256', secret)
        .update(payload)
        .digest('hex')
    return crypto.timingSafeEqual(
        Buffer.from(expected),
        Buffer.from(signature.replace('sha256=', ''))
    )
}

// PostPenguin webhook endpoint
app.post('/api/webhooks/postpenguin', async (req, res) => {
    try {
        // Verify signature
        const signature = req.headers['x-postpenguin-signature']
        const secret = process.env.POSTPENGUIN_WEBHOOK_SECRET
        
        if (secret && signature) {
            if (!verifySignature(JSON.stringify(req.body), signature, secret)) {
                return res.status(401).json({ error: 'Invalid signature' })
            }
        }
        
        const { 
            postPenguinId, 
            title, 
            slug, 
            html, 
            meta_title, 
            meta_description, 
            featured_image,
            tags 
        } = req.body
        
        if (!title || !slug || !html) {
            return res.status(400).json({ error: 'Missing required fields' })
        }
        
        const id = postPenguinId || `pp_${Date.now()}`
        const now = new Date()
        
        // Upsert post
        const [result] = await pool.execute(`
            INSERT INTO posts (
                postpenguin_id, title, slug, html, meta_title, 
                meta_description, featured_image, tags, status, published_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'publish', ?)
            ON DUPLICATE KEY UPDATE
                title = VALUES(title),
                slug = VALUES(slug),
                html = VALUES(html),
                meta_title = VALUES(meta_title),
                meta_description = VALUES(meta_description),
                featured_image = VALUES(featured_image),
                tags = VALUES(tags),
                published_at = VALUES(published_at)
        `, [
            id,
            title,
            slug,
            html,
            meta_title || title,
            meta_description || '',
            featured_image || null,
            JSON.stringify(tags || []),
            now
        ])
        
        res.json({
            success: true,
            postId: result.insertId || id,
            postPenguinId: id
        })
        
    } catch (error) {
        console.error('Webhook error:', error)
        res.status(500).json({ error: 'Internal server error' })
    }
})

// Fetch posts API
app.get('/api/posts', async (req, res) => {
    try {
        const { slug, limit = 10, offset = 0 } = req.query
        
        if (slug) {
            const [rows] = await pool.execute(
                'SELECT * FROM posts WHERE slug = ? AND status = ?',
                [slug, 'publish']
            )
            if (rows.length === 0) {
                return res.status(404).json({ error: 'Post not found' })
            }
            return res.json({ post: rows[0] })
        }
        
        const [posts] = await pool.execute(
            `SELECT * FROM posts WHERE status = 'publish' 
             ORDER BY published_at DESC LIMIT ? OFFSET ?`,
            [parseInt(limit), parseInt(offset)]
        )
        
        const [[{ total }]] = await pool.execute(
            "SELECT COUNT(*) as total FROM posts WHERE status = 'publish'"
        )
        
        res.json({
            posts,
            pagination: { total, limit: parseInt(limit), offset: parseInt(offset) }
        })
        
    } catch (error) {
        console.error('Error:', error)
        res.status(500).json({ error: 'Internal server error' })
    }
})

const PORT = process.env.PORT || 3001
app.listen(PORT, () => console.log(`Server running on port ${PORT}`))

🐍 Python with MySQL Connector

# webhook_handler.py
from flask import Flask, request, jsonify
import mysql.connector
from mysql.connector import pooling
import hmac
import hashlib
import os
import json
from datetime import datetime

app = Flask(__name__)

# MySQL connection pool
db_config = {
    'host': os.environ.get('MYSQL_HOST', 'localhost'),
    'user': os.environ['MYSQL_USER'],
    'password': os.environ['MYSQL_PASSWORD'],
    'database': os.environ['MYSQL_DATABASE'],
}

pool = pooling.MySQLConnectionPool(pool_name="mypool", pool_size=5, **db_config)

def verify_signature(payload, signature, secret):
    expected = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature.replace('sha256=', ''))

@app.route('/api/webhooks/postpenguin', methods=['POST'])
def webhook():
    try:
        signature = request.headers.get('X-PostPenguin-Signature', '')
        secret = os.environ.get('POSTPENGUIN_WEBHOOK_SECRET', '')
        
        if secret and signature:
            if not verify_signature(request.get_data(as_text=True), signature, secret):
                return jsonify({'error': 'Invalid signature'}), 401
        
        data = request.json
        
        if not all(k in data for k in ['title', 'slug', 'html']):
            return jsonify({'error': 'Missing required fields'}), 400
        
        conn = pool.get_connection()
        cursor = conn.cursor()
        
        post_id = data.get('postPenguinId', f'pp_{int(datetime.now().timestamp())}')
        now = datetime.now()
        
        cursor.execute('''
            INSERT INTO posts (
                postpenguin_id, title, slug, html, meta_title,
                meta_description, featured_image, tags, status, published_at
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, 'publish', %s)
            ON DUPLICATE KEY UPDATE
                title = VALUES(title),
                slug = VALUES(slug),
                html = VALUES(html),
                meta_title = VALUES(meta_title),
                meta_description = VALUES(meta_description),
                featured_image = VALUES(featured_image),
                tags = VALUES(tags),
                published_at = VALUES(published_at)
        ''', (
            post_id,
            data['title'],
            data['slug'],
            data['html'],
            data.get('meta_title', data['title']),
            data.get('meta_description', ''),
            data.get('featured_image'),
            json.dumps(data.get('tags', [])),
            now
        ))
        
        conn.commit()
        cursor.close()
        conn.close()
        
        return jsonify({'success': True, 'postId': post_id})
        
    except Exception as e:
        print(f'Webhook error: {e}')
        return jsonify({'error': 'Internal server error'}), 500

if __name__ == '__main__':
    app.run(port=3001)

🐘 PHP with PDO

<?php
// webhook.php

$host = getenv('MYSQL_HOST') ?: 'localhost';
$dbname = getenv('MYSQL_DATABASE');
$user = getenv('MYSQL_USER');
$pass = getenv('MYSQL_PASSWORD');

try {
    $pdo = new PDO("mysql:host=$host;dbname=$dbname;charset=utf8mb4", $user, $pass);
    $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
} catch (PDOException $e) {
    http_response_code(500);
    echo json_encode(['error' => 'Database connection failed']);
    exit;
}

// Verify signature
$signature = $_SERVER['HTTP_X_POSTPENGUIN_SIGNATURE'] ?? '';
$secret = getenv('POSTPENGUIN_WEBHOOK_SECRET');
$payload = file_get_contents('php://input');

if ($secret && $signature) {
    $expected = hash_hmac('sha256', $payload, $secret);
    if (!hash_equals($expected, str_replace('sha256=', '', $signature))) {
        http_response_code(401);
        echo json_encode(['error' => 'Invalid signature']);
        exit;
    }
}

$data = json_decode($payload, true);

if (!isset($data['title'], $data['slug'], $data['html'])) {
    http_response_code(400);
    echo json_encode(['error' => 'Missing required fields']);
    exit;
}

$postId = $data['postPenguinId'] ?? 'pp_' . time();

$stmt = $pdo->prepare('
    INSERT INTO posts (
        postpenguin_id, title, slug, html, meta_title,
        meta_description, featured_image, tags, status, published_at
    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, "publish", NOW())
    ON DUPLICATE KEY UPDATE
        title = VALUES(title),
        slug = VALUES(slug),
        html = VALUES(html),
        meta_title = VALUES(meta_title),
        meta_description = VALUES(meta_description),
        featured_image = VALUES(featured_image),
        tags = VALUES(tags),
        published_at = VALUES(published_at)
');

$stmt->execute([
    $postId,
    $data['title'],
    $data['slug'],
    $data['html'],
    $data['meta_title'] ?? $data['title'],
    $data['meta_description'] ?? '',
    $data['featured_image'] ?? null,
    json_encode($data['tags'] ?? [])
]);

header('Content-Type: application/json');
echo json_encode([
    'success' => true,
    'postId' => $postId
]);

⚙️ Environment Variables

# .env
MYSQL_HOST=localhost
MYSQL_USER=your_username
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=your_database
POSTPENGUIN_WEBHOOK_SECRET=your-secret-key-here

🧪 Testing

curl -X POST http://localhost:3001/api/webhooks/postpenguin \
  -H "Content-Type: application/json" \
  -d '{
    "postPenguinId": "test_mysql_123",
    "title": "Test MySQL Post",
    "slug": "test-mysql-post",
    "html": "<p className="text-gray-700">This is a test post stored in MySQL.</p>",
    "meta_title": "Test Post",
    "meta_description": "Testing MySQL webhook",
    "tags": ["test", "mysql"]
  }'

Verify in MySQL

mysql -u your_user -p your_database

SELECT * FROM posts WHERE slug = 'test-mysql-post';
SELECT id, title, slug, published_at FROM posts ORDER BY published_at DESC LIMIT 5;