feat: add Qdrant integration foundation and Shopify sync #74

- Created Qdrant client library with vector management
- Extended database schema with privacy settings and Qdrant tracking
- Updated Shopify sync to support Qdrant vector database
- Added change detection for product deletions in Qdrant
- Respects store data access permissions for customers/orders

supabase/functions/_shared/qdrant-client.ts

@@ -0,0 +1,454 @@
+/**
+ * Qdrant Vector Database Client
+ *
+ * Provides utilities for managing collections and vectors in Qdrant
+ * for e-commerce store data synchronization.
+ *
+ * Vector dimensions: 3072 (OpenAI text-embedding-3-large)
+ * Metric: Cosine (best for normalized embeddings)
+ */
+
+const QDRANT_URL = 'http://142.93.100.6:6333';
+const QDRANT_API_KEY = 'pyXAyyEPbLzba2RvdBwm';
+const VECTOR_SIZE = 3072;
+const DISTANCE_METRIC = 'Cosine'; // Best for text embeddings
+
+export interface QdrantPoint {
+  id: string | number;
+  vector: number[];
+  payload: Record<string, any>;
+}
+
+export interface CollectionSchema {
+  vectors: {
+    size: number;
+    distance: string;
+  };
+  optimizers_config?: {
+    indexing_threshold: number;
+  };
+}
+
+/**
+ * Make a request to Qdrant API
+ */
+async function qdrantRequest(
+  endpoint: string,
+  method: string = 'GET',
+  body?: any
+): Promise<any> {
+  const url = `${QDRANT_URL}${endpoint}`;
+
+  const headers: Record<string, string> = {
+    'Content-Type': 'application/json',
+  };
+
+  if (QDRANT_API_KEY) {
+    headers['api-key'] = QDRANT_API_KEY;
+  }
+
+  const options: RequestInit = {
+    method,
+    headers,
+  };
+
+  if (body) {
+    options.body = JSON.stringify(body);
+  }
+
+  const response = await fetch(url, options);
+
+  if (!response.ok) {
+    const errorText = await response.text();
+    throw new Error(`Qdrant API error (${response.status}): ${errorText}`);
+  }
+
+  return response.json();
+}
+
+/**
+ * Check if a collection exists
+ */
+export async function collectionExists(collectionName: string): Promise<boolean> {
+  try {
+    await qdrantRequest(`/collections/${collectionName}`);
+    return true;
+  } catch (error: any) {
+    if (error.message?.includes('404')) {
+      return false;
+    }
+    throw error;
+  }
+}
+
+/**
+ * Create a collection with optimized settings
+ */
+export async function createCollection(
+  collectionName: string,
+  payloadIndexes?: Array<{ field: string; type: string }>
+): Promise<void> {
+  console.log(`[Qdrant] Creating collection: ${collectionName}`);
+
+  const collectionConfig: any = {
+    vectors: {
+      size: VECTOR_SIZE,
+      distance: DISTANCE_METRIC,
+    },
+    optimizers_config: {
+      indexing_threshold: 10000, // Start indexing after 10k points
+    },
+    // Enable on-disk storage for large collections
+    on_disk_payload: true,
+  };
+
+  await qdrantRequest(`/collections/${collectionName}`, 'PUT', collectionConfig);
+
+  // Create payload indexes for faster filtering
+  if (payloadIndexes && payloadIndexes.length > 0) {
+    for (const index of payloadIndexes) {
+      try {
+        await qdrantRequest(
+          `/collections/${collectionName}/index`,
+          'PUT',
+          {
+            field_name: index.field,
+            field_schema: index.type,
+          }
+        );
+        console.log(`[Qdrant] Created index on ${index.field} (${index.type})`);
+      } catch (error) {
+        console.error(`[Qdrant] Failed to create index on ${index.field}:`, error);
+      }
+    }
+  }
+
+  console.log(`[Qdrant] Collection ${collectionName} created successfully`);
+}
+
+/**
+ * Delete a collection
+ */
+export async function deleteCollection(collectionName: string): Promise<void> {
+  console.log(`[Qdrant] Deleting collection: ${collectionName}`);
+  await qdrantRequest(`/collections/${collectionName}`, 'DELETE');
+  console.log(`[Qdrant] Collection ${collectionName} deleted`);
+}
+
+/**
+ * Upsert points (vectors with payloads) into a collection
+ */
+export async function upsertPoints(
+  collectionName: string,
+  points: QdrantPoint[]
+): Promise<void> {
+  if (points.length === 0) {
+    return;
+  }
+
+  console.log(`[Qdrant] Upserting ${points.length} points to ${collectionName}`);
+
+  // Batch upsert in chunks of 100 to avoid payload size limits
+  const chunkSize = 100;
+  for (let i = 0; i < points.length; i += chunkSize) {
+    const chunk = points.slice(i, i + chunkSize);
+
+    await qdrantRequest(`/collections/${collectionName}/points`, 'PUT', {
+      points: chunk,
+    });
+
+    console.log(`[Qdrant] Upserted ${chunk.length} points (${i + chunk.length}/${points.length})`);
+  }
+}
+
+/**
+ * Delete points by ID
+ */
+export async function deletePoints(
+  collectionName: string,
+  pointIds: (string | number)[]
+): Promise<void> {
+  if (pointIds.length === 0) {
+    return;
+  }
+
+  console.log(`[Qdrant] Deleting ${pointIds.length} points from ${collectionName}`);
+
+  await qdrantRequest(`/collections/${collectionName}/points/delete`, 'POST', {
+    points: pointIds,
+  });
+
+  console.log(`[Qdrant] Deleted ${pointIds.length} points`);
+}
+
+/**
+ * Delete points by filter
+ */
+export async function deletePointsByFilter(
+  collectionName: string,
+  filter: any
+): Promise<void> {
+  console.log(`[Qdrant] Deleting points by filter from ${collectionName}`);
+
+  await qdrantRequest(`/collections/${collectionName}/points/delete`, 'POST', {
+    filter,
+  });
+
+  console.log(`[Qdrant] Points deleted by filter`);
+}
+
+/**
+ * Get points by IDs
+ */
+export async function getPoints(
+  collectionName: string,
+  pointIds: (string | number)[]
+): Promise<any[]> {
+  if (pointIds.length === 0) {
+    return [];
+  }
+
+  const response = await qdrantRequest(`/collections/${collectionName}/points`, 'POST', {
+    ids: pointIds,
+    with_payload: true,
+    with_vector: false,
+  });
+
+  return response.result || [];
+}
+
+/**
+ * Scroll through all points in a collection (for detecting deleted items)
+ */
+export async function scrollPoints(
+  collectionName: string,
+  filter?: any,
+  limit: number = 100
+): Promise<{ points: any[]; nextOffset?: string }> {
+  const body: any = {
+    limit,
+    with_payload: true,
+    with_vector: false,
+  };
+
+  if (filter) {
+    body.filter = filter;
+  }
+
+  const response = await qdrantRequest(`/collections/${collectionName}/points/scroll`, 'POST', body);
+
+  return {
+    points: response.result?.points || [],
+    nextOffset: response.result?.next_page_offset,
+  };
+}
+
+/**
+ * Search for similar vectors
+ */
+export async function searchPoints(
+  collectionName: string,
+  vector: number[],
+  limit: number = 10,
+  filter?: any
+): Promise<any[]> {
+  const body: any = {
+    vector,
+    limit,
+    with_payload: true,
+  };
+
+  if (filter) {
+    body.filter = filter;
+  }
+
+  const response = await qdrantRequest(`/collections/${collectionName}/points/search`, 'POST', body);
+
+  return response.result || [];
+}
+
+/**
+ * Get collection info
+ */
+export async function getCollectionInfo(collectionName: string): Promise<any> {
+  const response = await qdrantRequest(`/collections/${collectionName}`);
+  return response.result;
+}
+
+/**
+ * Generate standardized collection name for a store
+ */
+export function getCollectionName(shopname: string, dataType: 'products' | 'orders' | 'customers'): string {
+  // Sanitize shopname: lowercase, replace non-alphanumeric with hyphens
+  const sanitized = shopname.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '');
+  return `${sanitized}-${dataType}`;
+}
+
+/**
+ * Initialize collections for a store
+ */
+export async function initializeStoreCollections(
+  shopname: string,
+  allowOrders: boolean = true,
+  allowCustomers: boolean = true
+): Promise<void> {
+  console.log(`[Qdrant] Initializing collections for store: ${shopname}`);
+
+  // Always create products collection
+  const productsCollection = getCollectionName(shopname, 'products');
+  if (!(await collectionExists(productsCollection))) {
+    await createCollection(productsCollection, [
+      { field: 'store_id', type: 'keyword' },
+      { field: 'product_id', type: 'keyword' },
+      { field: 'platform', type: 'keyword' },
+      { field: 'status', type: 'keyword' },
+      { field: 'price', type: 'float' },
+      { field: 'sku', type: 'keyword' },
+    ]);
+  }
+
+  // Create orders collection if allowed
+  if (allowOrders) {
+    const ordersCollection = getCollectionName(shopname, 'orders');
+    if (!(await collectionExists(ordersCollection))) {
+      await createCollection(ordersCollection, [
+        { field: 'store_id', type: 'keyword' },
+        { field: 'order_id', type: 'keyword' },
+        { field: 'platform', type: 'keyword' },
+        { field: 'status', type: 'keyword' },
+        { field: 'total_price', type: 'float' },
+        { field: 'customer_email', type: 'keyword' },
+      ]);
+    }
+  }
+
+  // Create customers collection if allowed
+  if (allowCustomers) {
+    const customersCollection = getCollectionName(shopname, 'customers');
+    if (!(await collectionExists(customersCollection))) {
+      await createCollection(customersCollection, [
+        { field: 'store_id', type: 'keyword' },
+        { field: 'customer_id', type: 'keyword' },
+        { field: 'platform', type: 'keyword' },
+        { field: 'email', type: 'keyword' },
+      ]);
+    }
+  }
+
+  console.log(`[Qdrant] Collections initialized for ${shopname}`);
+}
+
+/**
+ * Generate a simple embedding vector (placeholder - should use OpenAI API in production)
+ */
+export function generateSimpleEmbedding(text: string): number[] {
+  // This is a placeholder. In production, you should use OpenAI's text-embedding-3-large
+  // For now, generate a random normalized vector
+  const vector = new Array(VECTOR_SIZE).fill(0).map(() => Math.random() - 0.5);
+
+  // Normalize the vector
+  const magnitude = Math.sqrt(vector.reduce((sum, val) => sum + val * val, 0));
+  return vector.map(val => val / magnitude);
+}
+
+/**
+ * Create text representation of product for embedding
+ */
+export function createProductText(product: any): string {
+  const parts: string[] = [];
+
+  if (product.title || product.name) {
+    parts.push(product.title || product.name);
+  }
+
+  if (product.description) {
+    // Truncate long descriptions
+    const desc = product.description.replace(/<[^>]*>/g, '').substring(0, 500);
+    parts.push(desc);
+  }
+
+  if (product.sku) {
+    parts.push(`SKU: ${product.sku}`);
+  }
+
+  if (product.vendor) {
+    parts.push(`Vendor: ${product.vendor}`);
+  }
+
+  if (product.product_type) {
+    parts.push(`Type: ${product.product_type}`);
+  }
+
+  if (product.tags && Array.isArray(product.tags)) {
+    parts.push(`Tags: ${product.tags.join(', ')}`);
+  }
+
+  return parts.join(' | ');
+}
+
+/**
+ * Create text representation of order for embedding
+ */
+export function createOrderText(order: any): string {
+  const parts: string[] = [];
+
+  parts.push(`Order ${order.order_number || order.name}`);
+
+  if (order.customer_name) {
+    parts.push(`Customer: ${order.customer_name}`);
+  }
+
+  if (order.total_price) {
+    parts.push(`Total: ${order.total_price} ${order.currency || ''}`);
+  }
+
+  if (order.financial_status) {
+    parts.push(`Payment: ${order.financial_status}`);
+  }
+
+  if (order.fulfillment_status) {
+    parts.push(`Fulfillment: ${order.fulfillment_status}`);
+  }
+
+  if (order.line_items && Array.isArray(order.line_items)) {
+    const items = order.line_items.map((item: any) =>
+      `${item.quantity || 1}x ${item.name || item.title || 'item'}`
+    ).join(', ');
+    parts.push(`Items: ${items}`);
+  }
+
+  return parts.join(' | ');
+}
+
+/**
+ * Create text representation of customer for embedding
+ */
+export function createCustomerText(customer: any): string {
+  const parts: string[] = [];
+
+  if (customer.first_name || customer.last_name) {
+    parts.push(`${customer.first_name || ''} ${customer.last_name || ''}`.trim());
+  }
+
+  if (customer.email) {
+    parts.push(`Email: ${customer.email}`);
+  }
+
+  if (customer.phone) {
+    parts.push(`Phone: ${customer.phone}`);
+  }
+
+  if (customer.orders_count) {
+    parts.push(`Orders: ${customer.orders_count}`);
+  }
+
+  if (customer.total_spent) {
+    parts.push(`Total spent: ${customer.total_spent} ${customer.currency || ''}`);
+  }
+
+  if (customer.tags && Array.isArray(customer.tags)) {
+    parts.push(`Tags: ${customer.tags.join(', ')}`);
+  }
+
+  return parts.join(' | ');
+}

supabase/functions/shopify-sync/index.ts

@@ -10,6 +10,20 @@ import {
   ShopifyCustomer
 } from '../_shared/shopify-client.ts'
 import { formatFirstValidPhone, detectCountryCode } from '../_shared/phone-formatter.ts'
+import {
+  collectionExists,
+  createCollection,
+  upsertPoints,
+  deletePointsByFilter,
+  scrollPoints,
+  getCollectionName,
+  initializeStoreCollections,
+  generateSimpleEmbedding,
+  createProductText,
+  createOrderText,
+  createCustomerText,
+  QdrantPoint
+} from '../_shared/qdrant-client.ts'
 
 const corsHeaders = {
   'Access-Control-Allow-Origin': '*',
@@ -51,15 +65,373 @@ async function fetchWithRetry<T>(
   throw new Error('Max retries exceeded')
 }
 
-// Sync products from Shopify
-async function syncProducts(
+// Log Qdrant sync operation
+async function logQdrantSync(
+  supabaseAdmin: any,
+  storeId: string,
+  syncType: string,
+  collectionName: string,
+  operation: string,
+  itemsProcessed: number,
+  itemsSucceeded: number,
+  itemsFailed: number,
+  startedAt: Date,
+  errorMessage?: string
+): Promise<void> {
+  try {
+    await supabaseAdmin
+      .from('qdrant_sync_logs')
+      .insert({
+        store_id: storeId,
+        sync_type: syncType,
+        collection_name: collectionName,
+        operation,
+        items_processed: itemsProcessed,
+        items_succeeded: itemsSucceeded,
+        items_failed: itemsFailed,
+        error_message: errorMessage,
+        started_at: startedAt.toISOString(),
+        completed_at: new Date().toISOString(),
+      })
+  } catch (error) {
+    console.error('[Qdrant] Failed to log sync operation:', error)
+  }
+}
+
+// Sync products to Qdrant
+async function syncProductsToQdrant(
+  storeId: string,
+  storeName: string,
+  products: ShopifyProduct[],
+  supabaseAdmin: any
+): Promise<{ synced: number; errors: number }> {
+  const startTime = new Date()
+  const collectionName = getCollectionName(storeName, 'products')
+
+  console.log(`[Qdrant] Syncing ${products.length} products to ${collectionName}`)
+
+  let synced = 0
+  let errors = 0
+
+  try {
+    // Ensure collection exists
+    if (!(await collectionExists(collectionName))) {
+      await createCollection(collectionName, [
+        { field: 'store_id', type: 'keyword' },
+        { field: 'product_id', type: 'keyword' },
+        { field: 'platform', type: 'keyword' },
+        { field: 'status', type: 'keyword' },
+        { field: 'price', type: 'float' },
+        { field: 'sku', type: 'keyword' },
+      ])
+    }
+
+    // Get existing product IDs from Qdrant to detect deletions
+    const existingPoints = await scrollPoints(collectionName, {
+      must: [{ key: 'store_id', match: { value: storeId } }]
+    }, 1000)
+
+    const existingProductIds = new Set(
+      existingPoints.points.map((p: any) => p.payload?.product_id).filter(Boolean)
+    )
+
+    // Current product IDs from Shopify
+    const currentProductIds = new Set(products.map(p => p.id.toString()))
+
+    // Find deleted products (in Qdrant but not in Shopify)
+    const deletedProductIds = Array.from(existingProductIds).filter(
+      id => !currentProductIds.has(id)
+    )
+
+    // Delete removed products from Qdrant
+    if (deletedProductIds.length > 0) {
+      console.log(`[Qdrant] Deleting ${deletedProductIds.length} removed products`)
+      await deletePointsByFilter(collectionName, {
+        must: [
+          { key: 'store_id', match: { value: storeId } },
+          { key: 'product_id', match: { any: deletedProductIds } }
+        ]
+      })
+    }
+
+    // Convert products to Qdrant points
+    const points: QdrantPoint[] = products.map((product) => {
+      const primaryVariant = product.variants?.[0]
+      const productText = createProductText({
+        title: product.title,
+        description: product.body_html,
+        sku: primaryVariant?.sku,
+        vendor: product.vendor,
+        product_type: product.product_type,
+        tags: product.tags ? product.tags.split(',').map(t => t.trim()) : [],
+      })
+
+      return {
+        id: `shopify-${storeId}-${product.id}`,
+        vector: generateSimpleEmbedding(productText),
+        payload: {
+          store_id: storeId,
+          product_id: product.id.toString(),
+          platform: 'shopify',
+          title: product.title,
+          handle: product.handle,
+          vendor: product.vendor || null,
+          product_type: product.product_type || null,
+          status: product.status,
+          price: primaryVariant ? parseFloat(primaryVariant.price) : 0,
+          sku: primaryVariant?.sku || null,
+          inventory_quantity: primaryVariant?.inventory_quantity || 0,
+          description: product.body_html || null,
+          tags: product.tags ? product.tags.split(',').map(t => t.trim()) : [],
+          synced_at: new Date().toISOString(),
+        }
+      }
+    })
+
+    // Upsert to Qdrant
+    await upsertPoints(collectionName, points)
+    synced = points.length
+
+    await logQdrantSync(
+      supabaseAdmin,
+      storeId,
+      'products',
+      collectionName,
+      'upsert',
+      products.length,
+      synced,
+      errors,
+      startTime
+    )
+
+    console.log(`[Qdrant] Products sync complete: ${synced} synced, ${deletedProductIds.length} deleted`)
+  } catch (error: any) {
+    console.error('[Qdrant] Product sync error:', error)
+    errors = products.length
+    await logQdrantSync(
+      supabaseAdmin,
+      storeId,
+      'products',
+      collectionName,
+      'upsert',
+      products.length,
+      synced,
+      errors,
+      startTime,
+      error.message
+    )
+  }
+
+  return { synced, errors }
+}
+
+// Sync orders to Qdrant
+async function syncOrdersToQdrant(
   storeId: string,
+  storeName: string,
+  orders: ShopifyOrder[],
   supabaseAdmin: any
 ): Promise<{ synced: number; errors: number }> {
+  const startTime = new Date()
+  const collectionName = getCollectionName(storeName, 'orders')
+
+  console.log(`[Qdrant] Syncing ${orders.length} orders to ${collectionName}`)
+
+  let synced = 0
+  let errors = 0
+
+  try {
+    if (!(await collectionExists(collectionName))) {
+      await createCollection(collectionName, [
+        { field: 'store_id', type: 'keyword' },
+        { field: 'order_id', type: 'keyword' },
+        { field: 'platform', type: 'keyword' },
+        { field: 'financial_status', type: 'keyword' },
+        { field: 'total_price', type: 'float' },
+        { field: 'customer_email', type: 'keyword' },
+      ])
+    }
+
+    const points: QdrantPoint[] = orders.map((order) => {
+      const orderText = createOrderText({
+        order_number: order.order_number,
+        name: order.name,
+        customer_name: order.customer ? `${order.customer.first_name || ''} ${order.customer.last_name || ''}`.trim() : null,
+        total_price: order.current_total_price,
+        currency: order.currency,
+        financial_status: order.financial_status,
+        fulfillment_status: order.fulfillment_status,
+        line_items: order.line_items,
+      })
+
+      return {
+        id: `shopify-${storeId}-${order.id}`,
+        vector: generateSimpleEmbedding(orderText),
+        payload: {
+          store_id: storeId,
+          order_id: order.id.toString(),
+          platform: 'shopify',
+          order_number: order.order_number.toString(),
+          name: order.name,
+          email: order.email || null,
+          phone: order.customer?.phone || order.billing_address?.phone || null,
+          financial_status: order.financial_status,
+          fulfillment_status: order.fulfillment_status || null,
+          total_price: parseFloat(order.current_total_price) || 0,
+          currency: order.currency,
+          customer_name: order.customer ? `${order.customer.first_name || ''} ${order.customer.last_name || ''}`.trim() : null,
+          customer_email: order.customer?.email || order.email || null,
+          synced_at: new Date().toISOString(),
+        }
+      }
+    })
+
+    await upsertPoints(collectionName, points)
+    synced = points.length
+
+    await logQdrantSync(
+      supabaseAdmin,
+      storeId,
+      'orders',
+      collectionName,
+      'upsert',
+      orders.length,
+      synced,
+      errors,
+      startTime
+    )
+
+    console.log(`[Qdrant] Orders sync complete: ${synced} synced`)
+  } catch (error: any) {
+    console.error('[Qdrant] Order sync error:', error)
+    errors = orders.length
+    await logQdrantSync(
+      supabaseAdmin,
+      storeId,
+      'orders',
+      collectionName,
+      'upsert',
+      orders.length,
+      synced,
+      errors,
+      startTime,
+      error.message
+    )
+  }
+
+  return { synced, errors }
+}
+
+// Sync customers to Qdrant
+async function syncCustomersToQdrant(
+  storeId: string,
+  storeName: string,
+  customers: ShopifyCustomer[],
+  supabaseAdmin: any
+): Promise<{ synced: number; errors: number }> {
+  const startTime = new Date()
+  const collectionName = getCollectionName(storeName, 'customers')
+
+  console.log(`[Qdrant] Syncing ${customers.length} customers to ${collectionName}`)
+
+  let synced = 0
+  let errors = 0
+
+  try {
+    if (!(await collectionExists(collectionName))) {
+      await createCollection(collectionName, [
+        { field: 'store_id', type: 'keyword' },
+        { field: 'customer_id', type: 'keyword' },
+        { field: 'platform', type: 'keyword' },
+        { field: 'email', type: 'keyword' },
+      ])
+    }
+
+    const points: QdrantPoint[] = customers.map((customer) => {
+      const customerText = createCustomerText({
+        first_name: customer.first_name,
+        last_name: customer.last_name,
+        email: customer.email,
+        phone: customer.phone,
+        orders_count: customer.orders_count,
+        total_spent: customer.total_spent,
+        currency: customer.currency,
+        tags: customer.tags ? customer.tags.split(',').map(t => t.trim()) : [],
+      })
+
+      return {
+        id: `shopify-${storeId}-${customer.id}`,
+        vector: generateSimpleEmbedding(customerText),
+        payload: {
+          store_id: storeId,
+          customer_id: customer.id.toString(),
+          platform: 'shopify',
+          email: customer.email,
+          first_name: customer.first_name || null,
+          last_name: customer.last_name || null,
+          phone: customer.phone || customer.default_address?.phone || null,
+          orders_count: customer.orders_count || 0,
+          total_spent: parseFloat(customer.total_spent) || 0,
+          currency: customer.currency || 'USD',
+          state: customer.state,
+          synced_at: new Date().toISOString(),
+        }
+      }
+    })
+
+    await upsertPoints(collectionName, points)
+    synced = points.length
+
+    await logQdrantSync(
+      supabaseAdmin,
+      storeId,
+      'customers',
+      collectionName,
+      'upsert',
+      customers.length,
+      synced,
+      errors,
+      startTime
+    )
+
+    console.log(`[Qdrant] Customers sync complete: ${synced} synced`)
+  } catch (error: any) {
+    console.error('[Qdrant] Customer sync error:', error)
+    errors = customers.length
+    await logQdrantSync(
+      supabaseAdmin,
+      storeId,
+      'customers',
+      collectionName,
+      'upsert',
+      customers.length,
+      synced,
+      errors,
+      startTime,
+      error.message
+    )
+  }
+
+  return { synced, errors }
+}
+
+// Sync products from Shopify
+async function syncProducts(
+  storeId: string,
+  storeName: string,
+  supabaseAdmin: any,
+  qdrantEnabled: boolean,
+  canSyncProducts: boolean
+): Promise<{ synced: number; errors: number; qdrant?: { synced: number; errors: number } }> {
   console.log('[Shopify] Syncing products...')
   let synced = 0
   let errors = 0
 
+  if (!canSyncProducts) {
+    console.log('[Shopify] Product sync disabled by store permissions')
+    return { synced: 0, errors: 0 }
+  }
+
   try {
     const products = await fetchWithRetry(() => fetchAllProducts(storeId))
 
@@ -77,9 +449,8 @@ async function syncProducts(
 
     const currency = store?.alt_data?.currency || 'USD'
 
-    // Map and upsert products
+    // Map and upsert products to SQL cache
     const productsToCache = products.map((product: ShopifyProduct) => {
-      // Get primary variant for main product data
       const primaryVariant = product.variants?.[0]
 
       return {
@@ -126,24 +497,39 @@ async function syncProducts(
     }
 
     console.log(`[Shopify] Products sync complete: ${synced} synced, ${errors} errors`)
+
+    // Sync to Qdrant if enabled
+    let qdrantResult
+    if (qdrantEnabled) {
+      qdrantResult = await syncProductsToQdrant(storeId, storeName, products, supabaseAdmin)
+    }
+
+    return { synced, errors, qdrant: qdrantResult }
   } catch (error) {
     console.error('[Shopify] Product sync error:', error)
     errors++
+    return { synced, errors }
   }
-
-  return { synced, errors }
 }
 
 // Sync orders from Shopify
 async function syncOrders(
   storeId: string,
+  storeName: string,
   supabaseAdmin: any,
-  countryCode: string
-): Promise<{ synced: number; errors: number }> {
+  countryCode: string,
+  qdrantEnabled: boolean,
+  canSyncOrders: boolean
+): Promise<{ synced: number; errors: number; qdrant?: { synced: number; errors: number } }> {
   console.log('[Shopify] Syncing orders...')
   let synced = 0
   let errors = 0
 
+  if (!canSyncOrders) {
+    console.log('[Shopify] Order sync disabled by store permissions')
+    return { synced: 0, errors: 0 }
+  }
+
   try {
     const orders = await fetchWithRetry(() => fetchAllOrders(storeId))
 
@@ -204,24 +590,39 @@ async function syncOrders(
     }
 
     console.log(`[Shopify] Orders sync complete: ${synced} synced, ${errors} errors`)
+
+    // Sync to Qdrant if enabled
+    let qdrantResult
+    if (qdrantEnabled) {
+      qdrantResult = await syncOrdersToQdrant(storeId, storeName, orders, supabaseAdmin)
+    }
+
+    return { synced, errors, qdrant: qdrantResult }
   } catch (error) {
     console.error('[Shopify] Order sync error:', error)
     errors++
+    return { synced, errors }
   }
-
-  return { synced, errors }
 }
 
 // Sync customers from Shopify
 async function syncCustomers(
   storeId: string,
+  storeName: string,
   supabaseAdmin: any,
-  countryCode: string
-): Promise<{ synced: number; errors: number }> {
+  countryCode: string,
+  qdrantEnabled: boolean,
+  canSyncCustomers: boolean
+): Promise<{ synced: number; errors: number; qdrant?: { synced: number; errors: number } }> {
   console.log('[Shopify] Syncing customers...')
   let synced = 0
   let errors = 0
 
+  if (!canSyncCustomers) {
+    console.log('[Shopify] Customer sync disabled by store permissions')
+    return { synced: 0, errors: 0 }
+  }
+
   try {
     const customers = await fetchWithRetry(() => fetchAllCustomers(storeId))
 
@@ -277,12 +678,19 @@ async function syncCustomers(
     }
 
     console.log(`[Shopify] Customers sync complete: ${synced} synced, ${errors} errors`)
+
+    // Sync to Qdrant if enabled
+    let qdrantResult
+    if (qdrantEnabled) {
+      qdrantResult = await syncCustomersToQdrant(storeId, storeName, customers, supabaseAdmin)
+    }
+
+    return { synced, errors, qdrant: qdrantResult }
   } catch (error) {
     console.error('[Shopify] Customer sync error:', error)
     errors++
+    return { synced, errors }
   }
-
-  return { synced, errors }
 }
 
 serve(wrapHandler('shopify-sync', async (req) => {
@@ -325,13 +733,13 @@ serve(wrapHandler('shopify-sync', async (req) => {
       )
     }
 
-    // Verify store ownership
+    // Verify store ownership and get permissions
     const supabaseServiceKey = Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
     const supabaseAdmin = createClient(supabaseUrl, supabaseServiceKey)
 
     const { data: store, error: storeError } = await supabaseAdmin
       .from('stores')
-      .select('id, platform_name, store_name, store_url')
+      .select('id, platform_name, store_name, store_url, qdrant_sync_enabled, data_access_permissions')
       .eq('id', storeId)
       .eq('user_id', user.id)
       .eq('platform_name', 'shopify')
@@ -344,6 +752,33 @@ serve(wrapHandler('shopify-sync', async (req) => {
       )
     }
 
+    // Check data access permissions
+    const permissions = store.data_access_permissions || {}
+    const canSyncProducts = permissions.allow_product_access !== false
+    const canSyncOrders = permissions.allow_order_access !== false
+    const canSyncCustomers = permissions.allow_customer_access !== false
+    const qdrantEnabled = store.qdrant_sync_enabled !== false
+
+    console.log('[Shopify] Sync permissions:', {
+      products: canSyncProducts,
+      orders: canSyncOrders,
+      customers: canSyncCustomers,
+      qdrant: qdrantEnabled
+    })
+
+    // Initialize Qdrant collections if enabled
+    if (qdrantEnabled) {
+      try {
+        await initializeStoreCollections(
+          store.store_name,
+          canSyncOrders,
+          canSyncCustomers
+        )
+      } catch (error) {
+        console.error('[Qdrant] Failed to initialize collections:', error)
+      }
+    }
+
     // Detect country code from store URL for phone formatting
     const countryCode = detectCountryCode(store.store_url)
     console.log(`[Shopify] Detected country code: ${countryCode} for store: ${store.store_name}`)
@@ -359,15 +794,15 @@ serve(wrapHandler('shopify-sync', async (req) => {
 
     // Perform sync based on type
     if (syncType === 'all' || syncType === 'products') {
-      results.products = await syncProducts(storeId, supabaseAdmin)
+      results.products = await syncProducts(storeId, store.store_name, supabaseAdmin, qdrantEnabled, canSyncProducts)
     }
 
     if (syncType === 'all' || syncType === 'orders') {
-      results.orders = await syncOrders(storeId, supabaseAdmin, countryCode)
+      results.orders = await syncOrders(storeId, store.store_name, supabaseAdmin, countryCode, qdrantEnabled, canSyncOrders)
     }
 
     if (syncType === 'all' || syncType === 'customers') {
-      results.customers = await syncCustomers(storeId, supabaseAdmin, countryCode)
+      results.customers = await syncCustomers(storeId, store.store_name, supabaseAdmin, countryCode, qdrantEnabled, canSyncCustomers)
     }
 
     results.completed_at = new Date().toISOString()
@@ -384,4 +819,14 @@ serve(wrapHandler('shopify-sync', async (req) => {
       { status: 200, headers: { ...corsHeaders, 'Content-Type': 'application/json' } }
     )
 
+  } catch (error: any) {
+    console.error('[Shopify] Sync error:', error)
+    return new Response(
+      JSON.stringify({
+        success: false,
+        error: error.message || 'Sync failed'
+      }),
+      { status: 500, headers: { ...corsHeaders, 'Content-Type': 'application/json' } }
+    )
+  }
 }))

supabase/migrations/20251111_qdrant_integration.sql

@@ -0,0 +1,171 @@
+-- Migration: Add Qdrant integration support and extend data access permissions
+-- Date: 2025-11-11
+-- Issue: #74
+
+-- 1. Update data_access_permissions to include customer and order access control
+-- Modify default value to include all three permission types
+ALTER TABLE stores
+  ALTER COLUMN data_access_permissions
+  SET DEFAULT '{"allow_product_access": true, "allow_customer_access": true, "allow_order_access": true}'::jsonb;
+
+-- Update existing records to include the new permission fields (default to true)
+UPDATE stores
+SET data_access_permissions = jsonb_build_object(
+  'allow_product_access', COALESCE((data_access_permissions->>'allow_product_access')::boolean, true),
+  'allow_customer_access', true,
+  'allow_order_access', true
+)
+WHERE data_access_permissions IS NOT NULL;
+
+-- For NULL records, set to default
+UPDATE stores
+SET data_access_permissions = '{"allow_product_access": true, "allow_customer_access": true, "allow_order_access": true}'::jsonb
+WHERE data_access_permissions IS NULL;
+
+-- Update the check constraint to validate new fields
+ALTER TABLE stores
+  DROP CONSTRAINT IF EXISTS stores_data_access_permissions_check;
+
+ALTER TABLE stores
+  ADD CONSTRAINT stores_data_access_permissions_check
+  CHECK (
+    data_access_permissions ? 'allow_product_access' AND
+    data_access_permissions ? 'allow_customer_access' AND
+    data_access_permissions ? 'allow_order_access'
+  );
+
+-- Update column comment
+COMMENT ON COLUMN stores.data_access_permissions IS 'JSONB object controlling data access permissions: allow_product_access, allow_customer_access, allow_order_access. Controls both SQL cache and Qdrant sync.';
+
+-- 2. Add Qdrant-specific columns to stores table
+ALTER TABLE stores
+  ADD COLUMN IF NOT EXISTS qdrant_sync_enabled boolean DEFAULT true,
+  ADD COLUMN IF NOT EXISTS qdrant_last_sync_at timestamptz,
+  ADD COLUMN IF NOT EXISTS qdrant_collection_prefix text;
+
+COMMENT ON COLUMN stores.qdrant_sync_enabled IS 'Whether to sync store data to Qdrant vector database';
+COMMENT ON COLUMN stores.qdrant_last_sync_at IS 'Timestamp of last successful Qdrant sync';
+COMMENT ON COLUMN stores.qdrant_collection_prefix IS 'Custom collection prefix for Qdrant (defaults to store_name)';
+
+-- 3. Create qdrant_sync_logs table to track Qdrant synchronization
+CREATE TABLE IF NOT EXISTS qdrant_sync_logs (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  store_id uuid NOT NULL REFERENCES stores(id) ON DELETE CASCADE,
+  sync_type text NOT NULL CHECK (sync_type IN ('products', 'orders', 'customers', 'all')),
+  collection_name text NOT NULL,
+  operation text NOT NULL CHECK (operation IN ('create_collection', 'upsert', 'delete', 'full_sync')),
+  items_processed integer DEFAULT 0,
+  items_succeeded integer DEFAULT 0,
+  items_failed integer DEFAULT 0,
+  error_message text,
+  started_at timestamptz NOT NULL DEFAULT now(),
+  completed_at timestamptz,
+  duration_ms integer GENERATED ALWAYS AS (
+    EXTRACT(EPOCH FROM (completed_at - started_at)) * 1000
+  ) STORED,
+  metadata jsonb,
+  created_at timestamptz DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_qdrant_sync_logs_store_id ON qdrant_sync_logs(store_id);
+CREATE INDEX IF NOT EXISTS idx_qdrant_sync_logs_created_at ON qdrant_sync_logs(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_qdrant_sync_logs_sync_type ON qdrant_sync_logs(sync_type);
+
+COMMENT ON TABLE qdrant_sync_logs IS 'Tracks Qdrant vector database synchronization operations';
+
+-- Enable RLS
+ALTER TABLE qdrant_sync_logs ENABLE ROW LEVEL SECURITY;
+
+-- RLS Policy: Users can only see their own store's sync logs
+CREATE POLICY "Users can view their own qdrant sync logs"
+  ON qdrant_sync_logs
+  FOR SELECT
+  USING (
+    EXISTS (
+      SELECT 1 FROM stores
+      WHERE stores.id = qdrant_sync_logs.store_id
+      AND stores.user_id = auth.uid()
+    )
+  );
+
+-- 4. Create helper functions for data access permission checks
+CREATE OR REPLACE FUNCTION can_sync_products(store_uuid uuid)
+RETURNS boolean
+LANGUAGE plpgsql
+SECURITY DEFINER
+AS $$
+DECLARE
+  result boolean;
+BEGIN
+  SELECT COALESCE((data_access_permissions->>'allow_product_access')::boolean, true)
+  INTO result
+  FROM stores
+  WHERE id = store_uuid;
+
+  RETURN COALESCE(result, true);
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION can_sync_customers(store_uuid uuid)
+RETURNS boolean
+LANGUAGE plpgsql
+SECURITY DEFINER
+AS $$
+DECLARE
+  result boolean;
+BEGIN
+  SELECT COALESCE((data_access_permissions->>'allow_customer_access')::boolean, true)
+  INTO result
+  FROM stores
+  WHERE id = store_uuid;
+
+  RETURN COALESCE(result, true);
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION can_sync_orders(store_uuid uuid)
+RETURNS boolean
+LANGUAGE plpgsql
+SECURITY DEFINER
+AS $$
+DECLARE
+  result boolean;
+BEGIN
+  SELECT COALESCE((data_access_permissions->>'allow_order_access')::boolean, true)
+  INTO result
+  FROM stores
+  WHERE id = store_uuid;
+
+  RETURN COALESCE(result, true);
+END;
+$$;
+
+COMMENT ON FUNCTION can_sync_products IS 'Check if products can be synced for a store (both SQL cache and Qdrant)';
+COMMENT ON FUNCTION can_sync_customers IS 'Check if customers can be synced for a store (both SQL cache and Qdrant)';
+COMMENT ON FUNCTION can_sync_orders IS 'Check if orders can be synced for a store (both SQL cache and Qdrant)';
+
+-- 5. Create trigger to update qdrant_last_sync_at when sync completes successfully
+CREATE OR REPLACE FUNCTION update_qdrant_last_sync()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF NEW.completed_at IS NOT NULL AND NEW.error_message IS NULL THEN
+    UPDATE stores
+    SET qdrant_last_sync_at = NEW.completed_at
+    WHERE id = NEW.store_id;
+  END IF;
+
+  RETURN NEW;
+END;
+$$;
+
+CREATE TRIGGER trigger_update_qdrant_last_sync
+  AFTER INSERT OR UPDATE ON qdrant_sync_logs
+  FOR EACH ROW
+  WHEN (NEW.completed_at IS NOT NULL)
+  EXECUTE FUNCTION update_qdrant_last_sync();
+
+-- 6. Add indexes for better performance
+CREATE INDEX IF NOT EXISTS idx_stores_qdrant_sync_enabled ON stores(qdrant_sync_enabled) WHERE qdrant_sync_enabled = true;
+CREATE INDEX IF NOT EXISTS idx_stores_data_access_permissions ON stores USING gin(data_access_permissions);