/*
 * $Id: MySQLIndexAccessor.java,v 1.19 2006/02/24 15:55:55 akabane Exp $
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 *
 */
package org.logical_paradox.koike.rss.index;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ResourceBundle;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.charset.CharsetUtils;
import org.logical_paradox.common.sql.DbUtils;
import org.logical_paradox.common.util.ExceptionUtils;
import org.logical_paradox.common.util.StringUtils;
import org.logical_paradox.koike.core.KoikeConstant;
import org.logical_paradox.koike.core.indexer.IndexAccessException;
import org.logical_paradox.koike.core.indexer.IndexAccessor;
import org.logical_paradox.koike.core.parser.Term;
import org.logical_paradox.koike.core.search.InvertedIndex;
import org.logical_paradox.koike.core.search.KoikeIndexResultSet;
import org.logical_paradox.koike.core.search.Location;
import org.logical_paradox.rss.rcm.accessor.mysql.validator.ConnectionValidation;
import org.logical_paradox.rss.rcm.accessor.mysql.validator.ConnectionValidator;

/**
 * mysql serverをストレージとするタイプのインデックスアクセサ
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.19 $
 */
public class MySQLIndexAccessor implements IndexAccessor, MySQLInterfaceConstant {
	/** ロガー */
	private static final Log log = LogFactory.getLog(MySQLIndexAccessor.class);
	/** インデックス削除SQL */
	private PreparedStatement removeStatement;
	/** インデックス検索SQL */
	private PreparedStatement searchStatement;
	/** データベース接続 */
	private Connection con;
	/** データベース接続監視スレッド */
	private ConnectionValidator validator;
	/** 転置インデックスレジストラ */
	private MySQLInvertedIndexRegistrar registrar;
	/** 索引語のエンコーディング */
	private String indexTermEncoding;
	/** 索引語検索時，エンコーディング変換を必要とするか(true:する / false:しない) */
	private boolean needToEncode = false;

	/** base64 encoder/decoder */
	private final Base64 base64 = new Base64();

	/**
	 * コンストラクタ．
	 * @param cv データベース接続監視スレッド
	 */
	public MySQLIndexAccessor(ConnectionValidator cv) throws Exception {
		validator = cv;
		init();
	}
	/**
	 * このインターフェースを初期化する．
	 * @throws Exception 初期化に失敗した(ドライバがロードできないなど)
	 */
	protected void init() throws Exception {
		ResourceBundle rb = ResourceBundle.getBundle("mysql");
		String jdbcDriverClass = rb.getString(PKEY_JDBC_DRIVER_CLASS);
		String connstr = rb.getString(PKEY_INDEX_CONNECT_STR);

		// データベース接続プーリングを作成する
		try {
			Class.forName(jdbcDriverClass);
			con = DriverManager.getConnection(connstr);

			String removeSql = rb.getString(PKEY_INDEX_REMOVE_SQL);
			String searchSql = rb.getString(PKEY_INDEX_SEARCH_SQL);

			removeStatement = con.prepareStatement(removeSql);
			searchStatement = con.prepareStatement(searchSql);
			
			// データベース接続監視の開始
			ConnectionValidation validation = new ConnectionValidation(con, "select count(*) from inverted_index");
			validator.add(validation);

			// 転置インデックスレジストラの作成
			registrar = new MySQLInvertedIndexRegistrar(con);
			registrar.start();

			// 索引語のエンコーディング変換が必要かどうかをテストする
			indexTermEncoding = rb.getString(PKEY_ENCODING_INDEX_TERM);
			if(StringUtils.isEmpty(indexTermEncoding) == false) {
				needToEncode = CharsetUtils.testNeedToEncode(indexTermEncoding);
			}
		} catch(Exception e) {
			// 何らかの原因により，初期化に失敗した場合はリソースの回収を行なう
			DbUtils.closeStatement(removeStatement);
			DbUtils.closeStatement(searchStatement);
			DbUtils.closeConnection(con);
			log.error("インデックスアクセスインターフェースの初期化に失敗:" + ExceptionUtils.stackTraceToString(e), e);
		}
	}
	/**
	 * このインターフェースが既にクローズされているかどうかを返す．
	 * @return true:クローズ済み / false:まだ開いている
	 */
	public boolean isClosed() {
		return con == null;
	}
	/**
	 * このインターフェースを閉じる．
	 * 外部リソースとの接続を安全確実にクローズする．
	 * @throws IndexAccessException 何らかの理由でクローズ処理に失敗した
	 */
	public void close() throws IndexAccessException {
		if(registrar != null) {
			registrar.shutdown();
			try {
				// スレッドの停止を待ち合わせ
				registrar.join();
			} catch(InterruptedException e) {
				// 停止処理が中断されたが動作には影響ないので無視する
			} finally {
				registrar = null;
			}
		}
	}

	/**
	 * 索引語を追加する．
	 * @param digest 文書を特定するダイジェストキー
	 * @param term 登録する索引語
	 * @throws IndexAccessException 何らかの理由で登録に失敗した
	 */
	public void addTerm(String digest, Term[] term) throws IndexAccessException {
		registrar.add(digest, term);
	}

	/**
	 * 指定索引語の位置情報を全て削除する．
	 * @param key 索引語
	 * @return 削除されたもの
	 */
	public Object removeTerm(String key) throws IndexAccessException {
		try {
			removeStatement.setString(1, key);
			removeStatement.executeUpdate();

			return null;
		} catch(SQLException se) {
			// SQLに失敗しても，そのまま無視する(プーリングしているため)
			log.warn("索引語の削除に失敗:" + ExceptionUtils.stackTraceToString(se), se);
			throw new IndexAccessException(se);
		}
	}
	/**
	 * 指定索引語の位置情報を検索する．
	 * @param key 索引語
	 * @return 検索結果(位置情報)
	 */
	public KoikeIndexResultSet getValue(String key) throws IndexAccessException {
		KoikeIndexResultSet kirs = new KoikeIndexResultSet();
		if(needToEncode) {
			try {
				key = CharsetUtils.encode(indexTermEncoding, key);
			} catch (Exception e) {
				// エンコーディング変換になぜか失敗
				log.warn("索引語[" + key + "]のエンコーディング変換に失敗しました:" + ExceptionUtils.stackTraceToString(e));
				throw new IndexAccessException(e);
			}
		}
		double begin = System.currentTimeMillis();
		ResultSet rs = null;
		try {
			byte[] b64encoded = null;
			if(needToEncode) {
				b64encoded = base64.encode(key.getBytes("ISO-8859-1"));
			} else {
				b64encoded = base64.encode(key.getBytes());
			}
			String term = new String(b64encoded, "ISO-8859-1");		// base64encodingはUS-ASCIIであるはず
			searchStatement.setString(1, term);
			rs = searchStatement.executeQuery();

			int cnt = 0;
			while(rs.next()) {
				cnt++;
				long docno = rs.getLong("docno");
				String location = rs.getString("location");
				if(StringUtils.isEmpty(location)) {
					continue;
				}
				// 転置インデックスを生成して返す．
				kirs = createInvertedIndex(kirs, (int)docno, location); 
			}

			log.trace("term [" + key + "] is found in [" + cnt + "] rows.");
		} catch (Exception e) {
			log.warn("索引語の検索に失敗:" + ExceptionUtils.stackTraceToString(e), e);
			throw new IndexAccessException(e);
		} finally {
			// リソースの回収(DB接続とステートメントキャッシュは再利用のためほっとく)
			DbUtils.closeResultSet(rs);
		}

		double fin = System.currentTimeMillis();
		log.trace("検索時間: " + ((fin-begin) / 1000.0) + "sec.");
		return kirs;
	}

	/**
	 * 文字列化された転置インデックスをオブジェクトに復元して返す．
	 * @param rs 途中までの検索結果(null:新規)
	 * @param docno 文書番号
	 * @param value 文字列化された位置情報
	 * @return 復元されたオブジェクト
	 */
	protected KoikeIndexResultSet createInvertedIndex(KoikeIndexResultSet rs, int docno, String value) {
		if(rs == null) {
			rs = new KoikeIndexResultSet();
		}
		String nodeId = "";

		// 索引語の位置情報はカンマ区切りである
		String[] values = value.split(",");
		for(int i = 0; i < values.length; i++) {
			int pos = Integer.parseInt(values[i]);												// 位置情報
			int blkno = (int)(docno / KoikeConstant.INVERTED_INDEX_BLK_SIZE);					// ブロック番号
			long documents = 1L << (long)(docno - (blkno * KoikeConstant.INVERTED_INDEX_BLK_SIZE));
			InvertedIndex iidx = rs.getInvertedIndex(nodeId, blkno);
			if(iidx != null) {
				documents = documents | iidx.getDocumentAvailability();
			} else {
				iidx = new InvertedIndex(nodeId, blkno, 0);
			}
			// document availabilityを再設定する(位置情報フィルタリングなし)
			iidx.setDocumentAvailability(documents, false);
			// 転置インデックスに対して位置情報を追加する
			iidx.addLocation(new Location(nodeId, docno, pos));

			rs.addInvertedIndex(iidx);
		}

		// 検索結果を返す
		return rs;
	}
}
