/*
 * Copyright (c) 2009, syuu
 * License under the NicoCache License.
 */

package com.dokukino.genkidama;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.net.HttpURLConnection;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

//import net.sbbi.upnp.impls.InternetGatewayDevice;

import ow.dht.ByteArray;
import ow.dht.DHT;
import ow.dht.DHTConfiguration;
import ow.dht.DHTFactory;
import ow.dht.ValueInfo;
import ow.id.ID;
import ow.routing.RoutingException;

import dareka.common.Config;
import dareka.common.Logger;
import dareka.processor.HttpResponseHeader;
import dareka.processor.URLResource;
import dareka.processor.impl.NicoApiUtil;

public class DHTManager {
	public static Log logger = LogFactory.getLog(DHTManager.class);
	public final static short APPLICATION_ID = (short) 0x0abc;
	public final static short APPLICATION_VERSION = 3;

	private boolean connected;
	DHT<String> dht;
	private static DHTManager instance = new DHTManager();
	String externalAddress;
	ByteArray hashedSecret;
	private int ringPort, dataPort;

	private DataServer dataServer;
	private Map<ID, String> distributedMap = new HashMap<ID, String>();
/*
	private InternetGatewayDevice igd;
*/
	
	private DHTManager() {
	}

	public static DHTManager getInstance() {
		return instance;
	}

	public int getRingPort() {
		return ringPort;
	}

	@SuppressWarnings("unchecked")
	public void connect() throws Exception {
		ringPort = Config.getInteger("dhtRingPort", 9899);
		dataPort = Config.getInteger("dhtDataPort", 9900);
		assert ringPort != dataPort;
		String secret = Config.getString("dhtSecret", null);
		assert secret != null;
		hashedSecret = new ByteArray(secret.getBytes("UTF-8")).hashWithSHA1();
		externalAddress = (String) XmlRpcInvoker.invoke("globalIp.probe", null);
/*
		if (Config.getBoolean("dhtUPnP", true)) {
			InternetGatewayDevice[] IGDs = InternetGatewayDevice
					.getDevices(5000);
			assert IGDs != null;
			igd = IGDs[0];
			LocalAddressLocator locator = new LocalAddressLocator(igd, Config
					.getString("dhtNetmask", "255.255.255.0"));
			String internalAddress = locator.locate().getHostAddress();
			assert igd.addPortMapping("Genkidama ringPort", null, ringPort,
					ringPort, internalAddress, 0, "TCP");
			assert igd.addPortMapping("Genkidama dataPort", null, dataPort,
					dataPort, internalAddress, 0, "TCP");
		}
*/
		DHTConfiguration config = DHTFactory.getDefaultConfiguration();
		config.setMessagingTransport("TCP");
		config.setRoutingAlgorithm("Kademlia");
		config.setRoutingStyle("Iterative");
		config.setDoUPnPNATTraversal(false);
		config.setSelfPort(ringPort);
		config.setSelfAddress(externalAddress);

		dht = DHTFactory.getDHT(APPLICATION_ID, APPLICATION_VERSION, config,
				null);

		joinLoop: while(true) {
			Object[] onlineUsers = (Object[]) XmlRpcInvoker.invoke(
					"onlineUsers.fetch", null);
			if (onlineUsers.length == 0) {
				if (logger.isInfoEnabled())logger.info("DHT started as standalone");
				break;
			}
			for (Object u : onlineUsers) {
				HashMap user = (HashMap) u;
				String address = (String) user.get("address");
				Integer port = (Integer) user.get("port");
				if (address.equals(externalAddress) && port == ringPort)
					continue;
				String url = address + ":" + port;
				if (logger.isInfoEnabled())logger.info("DHT joining to " + url);
				try {
					dht.joinOverlay(url);
					if (logger.isInfoEnabled())logger.info("DHT joined via " + url);
					break joinLoop;
				} catch (Exception e) {
					if (logger.isErrorEnabled())logger.error("Error Occurs:", e);
				}
			}
		}
		List params = new ArrayList();
		params.add(ringPort);
		params.add((int) APPLICATION_VERSION);
		XmlRpcInvoker.invoke("onlineUsers.register", params);

		dataServer = new DataServer();
		dataServer.bind(dataPort);
		connected = true;
	}

	public String getNicoCache(String id) throws RoutingException {
		if (!checkDeleted(id)) {
			ID key = ID.getSHA1BasedID(("nc:" + id).getBytes());
			logger.info("id: "+key);
			Set<ValueInfo<String>> uriList = dht.get(key);
			for (ValueInfo<String> v : uriList) {
				String uri = v.getValue();
				try {
					java.net.URL url = new java.net.URL(uri);
					if (url.getHost().equals(externalAddress))
						continue;
					HttpURLConnection conn = (HttpURLConnection) url
							.openConnection();
					conn.setRequestMethod("HEAD");
					conn.setConnectTimeout(Config
							.getInteger("dhtTimeout", 1000));
					conn.connect();
					conn.disconnect();
					if(logger.isInfoEnabled()) logger.info("DHT get: nc:" + id + " success");
					return uri;
				} catch (Exception e) {
					if(logger.isDebugEnabled()) logger.debug("Error Occurs:", e);
				}
			}
		}
		logger.info("DHT get: nc:" + id + " failed");
		return null;
	}

	@SuppressWarnings("unchecked")
	public void putNicoCache(String id) throws Exception {
		if (!checkDeleted(id)) {
			ID key = ID.getSHA1BasedID(("nc:" + id).getBytes());
			logger.info("DHT put: nc:" + id);
			logger.info("id: "+key);
			String uri = "http://" + externalAddress + ":" + dataPort + "/nc/"
					+ id;
			dht.setHashedSecretForPut(hashedSecret);
			distributedMap.put(key, uri);
			dht.put(key, uri);
			List params = new ArrayList();
			params.add(ringPort);
			params.add((int) APPLICATION_VERSION);
			params.add("nc:" + id);
			try {
				XmlRpcInvoker.invoke("onlineData.register", params);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				if (logger.isErrorEnabled())logger.error("Error Occurs:", e);
			}
		}
	}

	public static boolean checkDeleted(String id) {
		id = id.replaceAll("low$", "");
		String title = null;
		try {
			String url = NicoApiUtil.getThumbURL(id);
			URLResource r = new URLResource(url);
			// In the general case, proxy must let browser know a redirection,
			// but in this case, behave just as a client.
			r.setFollowRedirects(true);

			ByteArrayOutputStream bout = new ByteArrayOutputStream();
			r.transferTo(null, bout, null, null);

			ByteArrayInputStream bin = new ByteArrayInputStream(bout
					.toByteArray());
			new HttpResponseHeader(bin); // skip header

			title = NicoApiUtil.getThumbTitle(bin);

			bout.close();
			bin.close();
		} catch (Exception e) {
			Logger.error(e);
			e.printStackTrace();
		}
		if (title == null) {
			logger.info("DHT cacheDeleted: " + id + " -> true");
			return true;
		} else {
			logger.info("DHT cacheDeleted: " + id + " -> false");
			return false;
		}
	}

	@SuppressWarnings("unchecked")
	public void disconnect() throws RoutingException {
		for (Entry<ID, String> entry : distributedMap.entrySet()) {
			String key = entry.getValue().replaceFirst("^http://.+:[0-9]+/nc/",
					"nc:");
			if(logger.isInfoEnabled()) logger.info("DHT remove: " + key);
			List params = new ArrayList();
			params.add(ringPort);
			params.add((int) APPLICATION_VERSION);
			params.add(key);
			try {
				XmlRpcInvoker.invoke("onlineData.unregister", params);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				if (logger.isErrorEnabled()) logger.error("Error Occurs:", e);
			}
			dht.remove(entry.getKey(), hashedSecret);
		}
		List params = new ArrayList();
		params.add(ringPort);
		params.add((int) APPLICATION_VERSION);
		try {
			XmlRpcInvoker.invoke("onlineUsers.unregister", params);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			logger.error("Error Occurs:", e);
		}
		dht.stop();
		logger.info("DHT network disconnected");
		connected = false;
	}

	public boolean isConnected() {
		return connected;
	}
}
